max/ap-relay
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/max' into max

Browse Source
This commit is contained in:
Maxime Augier 2024-07-03 13:59:57 +02:00
commit 3bdffd6144
21 changed files with 432 additions and 429 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

488
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

21
Cargo.toml
View File

@ -1,11 +1,11 @@
[package] [package]
name = "ap-relay" name = "ap-relay"
description = "A simple activitypub relay" description = "A simple activitypub relay"
version = "0.3.106" version = "0.3.108"
authors = ["asonix <asonix@asonix.dog>"] authors = ["asonix <asonix@asonix.dog>"]
license = "AGPL-3.0" license = "AGPL-3.0"
readme = "README.md" readme = "README.md"
repository = "https://git.asonix.dog/asonix/ap-relay" repository = "https://git.asonix.dog/asonix/relay"
keywords = ["activitypub", "relay"] keywords = ["activitypub", "relay"]
edition = "2021" edition = "2021"
build = "src/build.rs" build = "src/build.rs"
@ -21,8 +21,7 @@ default = []
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies] [dependencies]
anyhow = "1.0" actix-web = { version = "4.4.0", default-features = false, features = ["compress-brotli", "compress-gzip", "rustls-0_22"] }
actix-web = { version = "4.4.0", default-features = false, features = ["compress-brotli", "compress-gzip", "rustls-0_21"] }
actix-webfinger = { version = "0.5.0", default-features = false } actix-webfinger = { version = "0.5.0", default-features = false }
activitystreams = "0.7.0-alpha.25" activitystreams = "0.7.0-alpha.25"
activitystreams-ext = "0.1.0-alpha.3" activitystreams-ext = "0.1.0-alpha.3"
@ -31,7 +30,8 @@ async-cpupool = "0.2.0"
bcrypt = "0.15" bcrypt = "0.15"
base64 = "0.21" base64 = "0.21"
clap = { version = "4.0.0", features = ["derive"] } clap = { version = "4.0.0", features = ["derive"] }
config = "0.13.0" color-eyre = "0.6.2"
config = { version = "0.14.0", default-features = false, features = ["toml", "json", "yaml"] }
console-subscriber = { version = "0.2", optional = true } console-subscriber = { version = "0.2", optional = true }
dashmap = "5.1.0" dashmap = "5.1.0"
dotenv = "0.15.0" dotenv = "0.15.0"
@ -57,11 +57,13 @@ reqwest-tracing = "0.4.5"
ring = "0.17.5" ring = "0.17.5"
rsa = { version = "0.9" } rsa = { version = "0.9" }
rsa-magic-public-key = "0.8.0" rsa-magic-public-key = "0.8.0"
rustls = "0.21.0" rustls = "0.22.0"
rustls-pemfile = "1.0.1" rustls-channel-resolver = "0.2.0"
rustls-pemfile = "2"
serde = { version = "1.0", features = ["derive"] } serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0" serde_json = "1.0"
sled = "0.34.7" sled = "0.34.7"
streem = "0.2.0"
teloxide = { version = "0.12.0", default-features = false, features = [ teloxide = { version = "0.12.0", default-features = false, features = [
"ctrlc_handler", "ctrlc_handler",
"macros", "macros",
@ -80,10 +82,9 @@ tracing-subscriber = { version = "0.3", features = [
] } ] }
tokio = { version = "1", features = ["full", "tracing"] } tokio = { version = "1", features = ["full", "tracing"] }
uuid = { version = "1", features = ["v4", "serde"] } uuid = { version = "1", features = ["v4", "serde"] }
streem = "0.2.0"
[dependencies.background-jobs] [dependencies.background-jobs]
version = "0.17.0" version = "0.18.0"
default-features = false default-features = false
features = ["error-logging", "metrics", "tokio"] features = ["error-logging", "metrics", "tokio"]
@ -101,7 +102,7 @@ features = ["middleware", "ring"]
version = "0.7.9" version = "0.7.9"
[build-dependencies] [build-dependencies]
anyhow = "1.0" color-eyre = "0.6.2"
dotenv = "0.15.0" dotenv = "0.15.0"
ructe = { version = "0.17.0", features = ["sass", "mime03"] } ructe = { version = "0.17.0", features = ["sass", "mime03"] }
toml = "0.8.0" toml = "0.8.0"

2
relay.nix
View File

@ -5,7 +5,7 @@
rustPlatform.buildRustPackage { rustPlatform.buildRustPackage {
pname = "relay"; pname = "relay";
version = "0.3.106"; version = "0.3.108";
src = ./.; src = ./.;
cargoLock.lockFile = ./Cargo.lock; cargoLock.lockFile = ./Cargo.lock;

4
src/build.rs
View File

@ -21,7 +21,7 @@ fn git_info() {
} }
} }
fn version_info() -> Result<(), anyhow::Error> { fn version_info() -> color_eyre::Result<()> {
let cargo_toml = Path::new(&std::env::var("CARGO_MANIFEST_DIR")?).join("Cargo.toml"); let cargo_toml = Path::new(&std::env::var("CARGO_MANIFEST_DIR")?).join("Cargo.toml");
let mut file = File::open(cargo_toml)?; let mut file = File::open(cargo_toml)?;
@ -42,7 +42,7 @@ fn version_info() -> Result<(), anyhow::Error> {
Ok(()) Ok(())
} }
fn main() -> Result<(), anyhow::Error> { fn main() -> color_eyre::Result<()> {
dotenv::dotenv().ok(); dotenv::dotenv().ok();
git_info(); git_info();

34
src/config.rs
View File

@ -12,9 +12,8 @@ use activitystreams::{
}; };
use config::Environment; use config::Environment;
use http_signature_normalization_actix::{digest::ring::Sha256, prelude::VerifyDigest}; use http_signature_normalization_actix::{digest::ring::Sha256, prelude::VerifyDigest};
use rustls::{Certificate, PrivateKey}; use rustls::sign::CertifiedKey;
use std::{ use std::{
io::BufReader,
net::{IpAddr, SocketAddr}, net::{IpAddr, SocketAddr},
path::PathBuf, path::PathBuf,
}; };
@ -312,43 +311,34 @@ impl Config {
Some((config.addr, config.port).into()) Some((config.addr, config.port).into())
} }
pub(crate) fn open_keys(&self) -> Result<Option<(Vec<Certificate>, PrivateKey)>, Error> { pub(crate) async fn open_keys(&self) -> Result<Option<CertifiedKey>, Error> {
let tls = if let Some(tls) = &self.tls { let tls = if let Some(tls) = &self.tls {
tls tls
} else { } else {
tracing::warn!("No TLS config present"); tracing::info!("No TLS config present");
return Ok(None); return Ok(None);
}; };
let mut certs_reader = BufReader::new(std::fs::File::open(&tls.cert)?); let certs_bytes = tokio::fs::read(&tls.cert).await?;
let certs = rustls_pemfile::certs(&mut certs_reader)?; let certs =
rustls_pemfile::certs(&mut certs_bytes.as_slice()).collect::<Result<Vec<_>, _>>()?;
if certs.is_empty() { if certs.is_empty() {
tracing::warn!("No certs read from certificate file"); tracing::warn!("No certs read from certificate file");
return Ok(None); return Ok(None);
} }
let mut key_reader = BufReader::new(std::fs::File::open(&tls.key)?); let key_bytes = tokio::fs::read(&tls.key).await?;
let key = rustls_pemfile::read_one(&mut key_reader)?; let key = if let Some(key) = rustls_pemfile::private_key(&mut key_bytes.as_slice())? {
key
let certs = certs.into_iter().map(Certificate).collect();
let key = if let Some(key) = key {
match key {
rustls_pemfile::Item::RSAKey(der) => PrivateKey(der),
rustls_pemfile::Item::PKCS8Key(der) => PrivateKey(der),
rustls_pemfile::Item::ECKey(der) => PrivateKey(der),
_ => {
tracing::warn!("Unknown key format: {:?}", key);
return Ok(None);
}
}
} else { } else {
tracing::warn!("Failed to read private key"); tracing::warn!("Failed to read private key");
return Ok(None); return Ok(None);
}; };
Ok(Some((certs, key))) let key = rustls::crypto::ring::sign::any_supported_type(&key)?;
Ok(Some(CertifiedKey::new(certs, key)))
} }
pub(crate) fn footer_blurb(&self) -> Option<crate::templates::Html<String>> { pub(crate) fn footer_blurb(&self) -> Option<crate::templates::Html<String>> {

104
src/error.rs
View File

@ -4,54 +4,82 @@ use actix_web::{
http::StatusCode, http::StatusCode,
HttpResponse, HttpResponse,
}; };
use background_jobs::BoxError;
use color_eyre::eyre::Error as Report;
use http_signature_normalization_reqwest::SignError; use http_signature_normalization_reqwest::SignError;
use std::{convert::Infallible, fmt::Debug, io}; use std::{convert::Infallible, io, sync::Arc};
use tokio::task::JoinError; use tokio::task::JoinError;
use tracing_error::SpanTrace;
#[derive(Clone)]
struct ArcKind {
kind: Arc<ErrorKind>,
}
impl std::fmt::Debug for ArcKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.kind.fmt(f)
}
}
impl std::fmt::Display for ArcKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.kind.fmt(f)
}
}
impl std::error::Error for ArcKind {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
self.kind.source()
}
}
pub(crate) struct Error { pub(crate) struct Error {
context: String, kind: ArcKind,
kind: ErrorKind, display: Box<str>,
debug: Box<str>,
} }
impl Error { impl Error {
fn kind(&self) -> &ErrorKind {
&self.kind.kind
}
pub(crate) fn is_breaker(&self) -> bool { pub(crate) fn is_breaker(&self) -> bool {
matches!(self.kind, ErrorKind::Breaker) matches!(self.kind(), ErrorKind::Breaker)
} }
pub(crate) fn is_not_found(&self) -> bool { pub(crate) fn is_not_found(&self) -> bool {
matches!(self.kind, ErrorKind::Status(_, StatusCode::NOT_FOUND)) matches!(self.kind(), ErrorKind::Status(_, StatusCode::NOT_FOUND))
} }
pub(crate) fn is_bad_request(&self) -> bool { pub(crate) fn is_bad_request(&self) -> bool {
matches!(self.kind, ErrorKind::Status(_, StatusCode::BAD_REQUEST)) matches!(self.kind(), ErrorKind::Status(_, StatusCode::BAD_REQUEST))
} }
pub(crate) fn is_gone(&self) -> bool { pub(crate) fn is_gone(&self) -> bool {
matches!(self.kind, ErrorKind::Status(_, StatusCode::GONE)) matches!(self.kind(), ErrorKind::Status(_, StatusCode::GONE))
} }
pub(crate) fn is_malformed_json(&self) -> bool { pub(crate) fn is_malformed_json(&self) -> bool {
matches!(self.kind, ErrorKind::Json(_)) matches!(self.kind(), ErrorKind::Json(_))
} }
} }
impl std::fmt::Debug for Error { impl std::fmt::Debug for Error {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
writeln!(f, "{:?}", self.kind) f.write_str(&self.debug)
} }
} }
impl std::fmt::Display for Error { impl std::fmt::Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
writeln!(f, "{}", self.kind)?; f.write_str(&self.display)
std::fmt::Display::fmt(&self.context, f)
} }
} }
impl std::error::Error for Error { impl std::error::Error for Error {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
self.kind.source() self.kind().source()
} }
} }
@ -60,25 +88,36 @@ where
ErrorKind: From<T>, ErrorKind: From<T>,
{ {
fn from(error: T) -> Self { fn from(error: T) -> Self {
let kind = ArcKind {
kind: Arc::new(ErrorKind::from(error)),
};
let report = Report::new(kind.clone());
let display = format!("{report}");
let debug = format!("{report:?}");
Error { Error {
context: SpanTrace::capture().to_string(), kind,
kind: error.into(), display: Box::from(display),
debug: Box::from(debug),
} }
} }
} }
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub(crate) enum ErrorKind { pub(crate) enum ErrorKind {
#[error("Error queueing job, {0}")] #[error("Error in extractor")]
Queue(anyhow::Error), Extractor(#[from] crate::extractors::ErrorKind),
#[error("Error in configuration, {0}")] #[error("Error queueing job")]
Queue(#[from] BoxError),
#[error("Error in configuration")]
Config(#[from] config::ConfigError), Config(#[from] config::ConfigError),
#[error("Couldn't parse key, {0}")] #[error("Couldn't parse key")]
Pkcs8(#[from] rsa::pkcs8::Error), Pkcs8(#[from] rsa::pkcs8::Error),
#[error("Couldn't encode public key, {0}")] #[error("Couldn't encode public key")]
Spki(#[from] rsa::pkcs8::spki::Error), Spki(#[from] rsa::pkcs8::spki::Error),
#[error("Couldn't sign request")] #[error("Couldn't sign request")]
@ -87,33 +126,36 @@ pub(crate) enum ErrorKind {
#[error("Couldn't make request")] #[error("Couldn't make request")]
Reqwest(#[from] reqwest::Error), Reqwest(#[from] reqwest::Error),
#[error("Couldn't build client")] #[error("Couldn't make request")]
ReqwestMiddleware(#[from] reqwest_middleware::Error), ReqwestMiddleware(#[from] reqwest_middleware::Error),
#[error("Couldn't parse IRI, {0}")] #[error("Couldn't parse IRI")]
ParseIri(#[from] activitystreams::iri_string::validate::Error), ParseIri(#[from] activitystreams::iri_string::validate::Error),
#[error("Couldn't normalize IRI, {0}")] #[error("Couldn't normalize IRI")]
NormalizeIri(#[from] std::collections::TryReserveError), NormalizeIri(#[from] std::collections::TryReserveError),
#[error("Couldn't perform IO, {0}")] #[error("Couldn't perform IO")]
Io(#[from] io::Error), Io(#[from] io::Error),
#[error("Couldn't sign string, {0}")] #[error("Couldn't sign string, {0}")]
Rsa(rsa::errors::Error), Rsa(rsa::errors::Error),
#[error("Couldn't use db, {0}")] #[error("Couldn't use db")]
Sled(#[from] sled::Error), Sled(#[from] sled::Error),
#[error("Couldn't do the json thing, {0}")] #[error("Couldn't do the json thing")]
Json(#[from] serde_json::Error), Json(#[from] serde_json::Error),
#[error("Couldn't sign request, {0}")] #[error("Couldn't sign request")]
Sign(#[from] SignError), Sign(#[from] SignError),
#[error("Couldn't sign digest")] #[error("Couldn't sign digest")]
Signature(#[from] rsa::signature::Error), Signature(#[from] rsa::signature::Error),
#[error("Couldn't prepare TLS private key")]
PrepareKey(#[from] rustls::Error),
#[error("Couldn't verify signature")] #[error("Couldn't verify signature")]
VerifySignature, VerifySignature,
@ -144,10 +186,10 @@ pub(crate) enum ErrorKind {
#[error("Wrong ActivityPub kind, {0}")] #[error("Wrong ActivityPub kind, {0}")]
Kind(String), Kind(String),
#[error("Too many CPUs, {0}")] #[error("Too many CPUs")]
CpuCount(#[from] std::num::TryFromIntError), CpuCount(#[from] std::num::TryFromIntError),
#[error("{0}")] #[error("Host mismatch")]
HostMismatch(#[from] CheckError), HostMismatch(#[from] CheckError),
#[error("Couldn't flush buffer")] #[error("Couldn't flush buffer")]
@ -201,7 +243,7 @@ pub(crate) enum ErrorKind {
impl ResponseError for Error { impl ResponseError for Error {
fn status_code(&self) -> StatusCode { fn status_code(&self) -> StatusCode {
match self.kind { match self.kind() {
ErrorKind::NotAllowed(_) | ErrorKind::WrongActor(_) | ErrorKind::BadActor(_, _) => { ErrorKind::NotAllowed(_) | ErrorKind::WrongActor(_) | ErrorKind::BadActor(_, _) => {
StatusCode::FORBIDDEN StatusCode::FORBIDDEN
} }
@ -221,7 +263,7 @@ impl ResponseError for Error {
.insert_header(("Content-Type", "application/activity+json")) .insert_header(("Content-Type", "application/activity+json"))
.body( .body(
serde_json::to_string(&serde_json::json!({ serde_json::to_string(&serde_json::json!({
"error": self.kind.to_string(), "error": self.kind().to_string(),
})) }))
.unwrap_or_else(|_| "{}".to_string()), .unwrap_or_else(|_| "{}".to_string()),
) )

74
src/extractors.rs
View File

@ -1,19 +1,15 @@
use actix_web::{ use actix_web::{
dev::Payload, dev::Payload,
error::ParseError, error::ParseError,
http::{ http::header::{from_one_raw_str, Header, HeaderName, HeaderValue, TryIntoHeaderValue},
header::{from_one_raw_str, Header, HeaderName, HeaderValue, TryIntoHeaderValue},
StatusCode,
},
web::Data, web::Data,
FromRequest, HttpMessage, HttpRequest, HttpResponse, ResponseError, FromRequest, HttpMessage, HttpRequest,
}; };
use bcrypt::{BcryptError, DEFAULT_COST}; use bcrypt::{BcryptError, DEFAULT_COST};
use http_signature_normalization_actix::{prelude::InvalidHeaderValue, Canceled, Spawn}; use http_signature_normalization_actix::{prelude::InvalidHeaderValue, Canceled, Spawn};
use std::{convert::Infallible, str::FromStr, time::Instant}; use std::{convert::Infallible, str::FromStr, time::Instant};
use tracing_error::SpanTrace;
use crate::{db::Db, future::LocalBoxFuture, spawner::Spawner}; use crate::{db::Db, error::Error, future::LocalBoxFuture, spawner::Spawner};
#[derive(Clone)] #[derive(Clone)]
pub(crate) struct AdminConfig { pub(crate) struct AdminConfig {
@ -83,74 +79,42 @@ impl Admin {
} }
} }
#[derive(Debug, thiserror::Error)]
#[error("Failed authentication")]
pub(crate) struct Error {
context: String,
#[source]
kind: ErrorKind,
}
impl Error { impl Error {
fn invalid() -> Self { fn invalid() -> Self {
Error { Error::from(ErrorKind::Invalid)
context: SpanTrace::capture().to_string(),
kind: ErrorKind::Invalid,
}
} }
fn missing_config() -> Self { fn missing_config() -> Self {
Error { Error::from(ErrorKind::MissingConfig)
context: SpanTrace::capture().to_string(),
kind: ErrorKind::MissingConfig,
}
} }
fn missing_db() -> Self { fn missing_db() -> Self {
Error { Error::from(ErrorKind::MissingDb)
context: SpanTrace::capture().to_string(),
kind: ErrorKind::MissingDb,
}
} }
fn missing_spawner() -> Self { fn missing_spawner() -> Self {
Error { Error::from(ErrorKind::MissingSpawner)
context: SpanTrace::capture().to_string(),
kind: ErrorKind::MissingSpawner,
}
} }
fn bcrypt_verify(e: BcryptError) -> Self { fn bcrypt_verify(e: BcryptError) -> Self {
Error { Error::from(ErrorKind::BCryptVerify(e))
context: SpanTrace::capture().to_string(),
kind: ErrorKind::BCryptVerify(e),
}
} }
fn bcrypt_hash(e: BcryptError) -> Self { fn bcrypt_hash(e: BcryptError) -> Self {
Error { Error::from(ErrorKind::BCryptHash(e))
context: SpanTrace::capture().to_string(),
kind: ErrorKind::BCryptHash(e),
}
} }
fn parse_header(e: ParseError) -> Self { fn parse_header(e: ParseError) -> Self {
Error { Error::from(ErrorKind::ParseHeader(e))
context: SpanTrace::capture().to_string(),
kind: ErrorKind::ParseHeader(e),
}
} }
fn canceled(_: Canceled) -> Self { fn canceled(_: Canceled) -> Self {
Error { Error::from(ErrorKind::Canceled)
context: SpanTrace::capture().to_string(),
kind: ErrorKind::Canceled,
}
} }
} }
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
enum ErrorKind { pub(crate) enum ErrorKind {
#[error("Invalid API Token")] #[error("Invalid API Token")]
Invalid, Invalid,
@ -176,20 +140,6 @@ enum ErrorKind {
ParseHeader(#[source] ParseError), ParseHeader(#[source] ParseError),
} }
impl ResponseError for Error {
fn status_code(&self) -> StatusCode {
match self.kind {
ErrorKind::Invalid | ErrorKind::ParseHeader(_) => StatusCode::BAD_REQUEST,
_ => StatusCode::INTERNAL_SERVER_ERROR,
}
}
fn error_response(&self) -> HttpResponse {
HttpResponse::build(self.status_code())
.json(serde_json::json!({ "msg": self.kind.to_string() }))
}
}
impl FromRequest for Admin { impl FromRequest for Admin {
type Error = Error; type Error = Error;
type Future = LocalBoxFuture<'static, Result<Self, Self::Error>>; type Future = LocalBoxFuture<'static, Result<Self, Self::Error>>;

5
src/jobs/apub/announce.rs
View File

@ -64,12 +64,13 @@ fn generate_announce(
impl Job for Announce { impl Job for Announce {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::apub::Announce"; const NAME: &'static str = "relay::jobs::apub::Announce";
const QUEUE: &'static str = "apub"; const QUEUE: &'static str = "apub";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/apub/follow.rs
View File

@ -113,12 +113,13 @@ fn generate_accept_follow(
impl Job for Follow { impl Job for Follow {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::apub::Follow"; const NAME: &'static str = "relay::jobs::apub::Follow";
const QUEUE: &'static str = "apub"; const QUEUE: &'static str = "apub";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/apub/forward.rs
View File

@ -49,12 +49,13 @@ impl Forward {
impl Job for Forward { impl Job for Forward {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::apub::Forward"; const NAME: &'static str = "relay::jobs::apub::Forward";
const QUEUE: &'static str = "apub"; const QUEUE: &'static str = "apub";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/apub/reject.rs
View File

@ -35,12 +35,13 @@ impl Reject {
impl Job for Reject { impl Job for Reject {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::apub::Reject"; const NAME: &'static str = "relay::jobs::apub::Reject";
const QUEUE: &'static str = "apub"; const QUEUE: &'static str = "apub";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/apub/undo.rs
View File

@ -50,12 +50,13 @@ impl Undo {
impl Job for Undo { impl Job for Undo {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::apub::Undo"; const NAME: &'static str = "relay::jobs::apub::Undo";
const QUEUE: &'static str = "apub"; const QUEUE: &'static str = "apub";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/contact.rs
View File

@ -87,13 +87,14 @@ fn to_contact(contact: AcceptedActors) -> Option<(String, String, IriString, Iri
impl Job for QueryContact { impl Job for QueryContact {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::QueryContact"; const NAME: &'static str = "relay::jobs::QueryContact";
const QUEUE: &'static str = "maintenance"; const QUEUE: &'static str = "maintenance";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

7
src/jobs/deliver.rs
View File

@ -35,7 +35,7 @@ impl Deliver {
} }
#[tracing::instrument(name = "Deliver", skip(state))] #[tracing::instrument(name = "Deliver", skip(state))]
async fn permform(self, state: JobState) -> Result<(), Error> { async fn perform(self, state: JobState) -> Result<(), Error> {
if let Err(e) = state if let Err(e) = state
.state .state
.requests .requests
@ -58,13 +58,14 @@ impl Deliver {
impl Job for Deliver { impl Job for Deliver {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::Deliver"; const NAME: &'static str = "relay::jobs::Deliver";
const QUEUE: &'static str = "deliver"; const QUEUE: &'static str = "deliver";
const BACKOFF: Backoff = Backoff::Exponential(8); const BACKOFF: Backoff = Backoff::Exponential(8);
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.permform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/deliver_many.rs
View File

@ -47,12 +47,13 @@ impl DeliverMany {
impl Job for DeliverMany { impl Job for DeliverMany {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::DeliverMany"; const NAME: &'static str = "relay::jobs::DeliverMany";
const QUEUE: &'static str = "deliver"; const QUEUE: &'static str = "deliver";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/instance.rs
View File

@ -167,13 +167,14 @@ impl QueryInstance {
impl Job for QueryInstance { impl Job for QueryInstance {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::QueryInstance"; const NAME: &'static str = "relay::jobs::QueryInstance";
const QUEUE: &'static str = "maintenance"; const QUEUE: &'static str = "maintenance";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/nodeinfo.rs
View File

@ -106,13 +106,14 @@ impl QueryNodeinfo {
impl Job for QueryNodeinfo { impl Job for QueryNodeinfo {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::QueryNodeinfo"; const NAME: &'static str = "relay::jobs::QueryNodeinfo";
const QUEUE: &'static str = "maintenance"; const QUEUE: &'static str = "maintenance";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/process_listeners.rs
View File

@ -25,12 +25,13 @@ impl Listeners {
impl Job for Listeners { impl Job for Listeners {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::Listeners"; const NAME: &'static str = "relay::jobs::Listeners";
const QUEUE: &'static str = "maintenance"; const QUEUE: &'static str = "maintenance";
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

5
src/jobs/record_last_online.rs
View File

@ -15,13 +15,14 @@ impl RecordLastOnline {
impl Job for RecordLastOnline { impl Job for RecordLastOnline {
type State = JobState; type State = JobState;
type Future = BoxFuture<'static, anyhow::Result<()>>; type Error = Error;
type Future = BoxFuture<'static, Result<(), Self::Error>>;
const NAME: &'static str = "relay::jobs::RecordLastOnline"; const NAME: &'static str = "relay::jobs::RecordLastOnline";
const QUEUE: &'static str = "maintenance"; const QUEUE: &'static str = "maintenance";
const BACKOFF: Backoff = Backoff::Linear(1); const BACKOFF: Backoff = Backoff::Linear(1);
fn run(self, state: Self::State) -> Self::Future { fn run(self, state: Self::State) -> Self::Future {
Box::pin(async move { self.perform(state).await.map_err(Into::into) }) Box::pin(self.perform(state))
} }
} }

70
src/main.rs
View File

@ -21,7 +21,7 @@ use tokio::task::JoinHandle;
use tracing_actix_web::TracingLogger; use tracing_actix_web::TracingLogger;
use tracing_error::ErrorLayer; use tracing_error::ErrorLayer;
use tracing_log::LogTracer; use tracing_log::LogTracer;
use tracing_subscriber::{filter::Targets, fmt::format::FmtSpan, layer::SubscriberExt, Layer}; use tracing_subscriber::{filter::Targets, layer::SubscriberExt, Layer};
mod admin; mod admin;
mod apub; mod apub;
@ -56,16 +56,15 @@ use self::{
fn init_subscriber( fn init_subscriber(
software_name: &'static str, software_name: &'static str,
opentelemetry_url: Option<&IriString>, opentelemetry_url: Option<&IriString>,
) -> Result<(), anyhow::Error> { ) -> color_eyre::Result<()> {
LogTracer::init()?; LogTracer::init()?;
color_eyre::install()?;
let targets: Targets = std::env::var("RUST_LOG") let targets: Targets = std::env::var("RUST_LOG")
.unwrap_or_else(|_| "warn,actix_web=debug,actix_server=debug,tracing_actix_web=info".into()) .unwrap_or_else(|_| "info".into())
.parse()?; .parse()?;
let format_layer = tracing_subscriber::fmt::layer() let format_layer = tracing_subscriber::fmt::layer().with_filter(targets.clone());
.with_span_events(FmtSpan::NEW | FmtSpan::CLOSE)
.with_filter(targets.clone());
#[cfg(feature = "console")] #[cfg(feature = "console")]
let console_layer = ConsoleLayer::builder() let console_layer = ConsoleLayer::builder()
@ -142,7 +141,7 @@ fn build_client(
} }
#[tokio::main] #[tokio::main]
async fn main() -> Result<(), anyhow::Error> { async fn main() -> color_eyre::Result<()> {
dotenv::dotenv().ok(); dotenv::dotenv().ok();
let config = Config::build()?; let config = Config::build()?;
@ -168,30 +167,30 @@ async fn main() -> Result<(), anyhow::Error> {
.add_recorder(recorder) .add_recorder(recorder)
.add_recorder(collector.clone()) .add_recorder(collector.clone())
.build(); .build();
metrics::set_global_recorder(recorder).map_err(|e| anyhow::anyhow!("{e}"))?; metrics::set_global_recorder(recorder).map_err(|e| color_eyre::eyre::eyre!("{e}"))?;
} else { } else {
collector.install()?; collector.install()?;
} }
tracing::warn!("Opening DB"); tracing::info!("Opening DB");
let db = Db::build(&config)?; let db = Db::build(&config)?;
tracing::warn!("Building caches"); tracing::info!("Building caches");
let actors = ActorCache::new(db.clone()); let actors = ActorCache::new(db.clone());
let media = MediaCache::new(db.clone()); let media = MediaCache::new(db.clone());
server_main(db, actors, media, collector, config).await?; server_main(db, actors, media, collector, config).await?;
tracing::warn!("Application exit"); tracing::info!("Application exit");
Ok(()) Ok(())
} }
fn client_main(config: Config, args: Args) -> JoinHandle<Result<(), anyhow::Error>> { fn client_main(config: Config, args: Args) -> JoinHandle<color_eyre::Result<()>> {
tokio::spawn(do_client_main(config, args)) tokio::spawn(do_client_main(config, args))
} }
async fn do_client_main(config: Config, args: Args) -> Result<(), anyhow::Error> { async fn do_client_main(config: Config, args: Args) -> color_eyre::Result<()> {
let client = build_client( let client = build_client(
&config.user_agent(), &config.user_agent(),
config.client_timeout(), config.client_timeout(),
@ -282,14 +281,14 @@ async fn server_main(
media: MediaCache, media: MediaCache,
collector: MemoryCollector, collector: MemoryCollector,
config: Config, config: Config,
) -> Result<(), anyhow::Error> { ) -> color_eyre::Result<()> {
let client = build_client( let client = build_client(
&config.user_agent(), &config.user_agent(),
config.client_timeout(), config.client_timeout(),
config.proxy_config(), config.proxy_config(),
)?; )?;
tracing::warn!("Creating state"); tracing::info!("Creating state");
let (signature_threads, verify_threads) = match config.signature_threads() { let (signature_threads, verify_threads) = match config.signature_threads() {
0 | 1 => (1, 1), 0 | 1 => (1, 1),
@ -309,15 +308,19 @@ async fn server_main(
let state = State::build(db.clone(), key_id, sign_spawner.clone(), client).await?; let state = State::build(db.clone(), key_id, sign_spawner.clone(), client).await?;
if let Some((token, admin_handle)) = config.telegram_info() { if let Some((token, admin_handle)) = config.telegram_info() {
tracing::warn!("Creating telegram handler"); tracing::info!("Creating telegram handler");
telegram::start(admin_handle.to_owned(), db.clone(), token); telegram::start(admin_handle.to_owned(), db.clone(), token);
} }
let keys = config.open_keys()?; let cert_resolver = config
.open_keys()
.await?
.map(rustls_channel_resolver::channel::<32>);
let bind_address = config.bind_address(); let bind_address = config.bind_address();
let sign_spawner2 = sign_spawner.clone(); let sign_spawner2 = sign_spawner.clone();
let verify_spawner2 = verify_spawner.clone(); let verify_spawner2 = verify_spawner.clone();
let config2 = config.clone();
let server = HttpServer::new(move || { let server = HttpServer::new(move || {
let job_server = let job_server =
create_workers(state.clone(), actors.clone(), media.clone(), config.clone()) create_workers(state.clone(), actors.clone(), media.clone(), config.clone())
@ -387,27 +390,42 @@ async fn server_main(
) )
}); });
if let Some((certs, key)) = keys { if let Some((cert_tx, cert_rx)) = cert_resolver {
tracing::warn!("Binding to {}:{} with TLS", bind_address.0, bind_address.1); let handle = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(30));
interval.tick().await;
loop {
interval.tick().await;
match config2.open_keys().await {
Ok(Some(key)) => cert_tx.update(key),
Ok(None) => tracing::warn!("Missing TLS keys"),
Err(e) => tracing::error!("Failed to read TLS keys {e}"),
}
}
});
tracing::info!("Binding to {}:{} with TLS", bind_address.0, bind_address.1);
let server_config = ServerConfig::builder() let server_config = ServerConfig::builder()
.with_safe_default_cipher_suites()
.with_safe_default_kx_groups()
.with_safe_default_protocol_versions()?
.with_no_client_auth() .with_no_client_auth()
.with_single_cert(certs, key)?; .with_cert_resolver(cert_rx);
server server
.bind_rustls_021(bind_address, server_config)? .bind_rustls_0_22(bind_address, server_config)?
.run() .run()
.await?; .await?;
handle.abort();
let _ = handle.await;
} else { } else {
tracing::warn!("Binding to {}:{}", bind_address.0, bind_address.1); tracing::info!("Binding to {}:{}", bind_address.0, bind_address.1);
server.bind(bind_address)?.run().await?; server.bind(bind_address)?.run().await?;
} }
sign_spawner2.close().await; sign_spawner2.close().await;
verify_spawner2.close().await; verify_spawner2.close().await;
tracing::warn!("Server closed"); tracing::info!("Server closed");
Ok(()) Ok(())
} }

2
src/spawner.rs
View File

@ -8,7 +8,7 @@ pub(crate) struct Spawner {
} }
impl Spawner { impl Spawner {
pub(crate) fn build(name: &'static str, threads: u16) -> anyhow::Result<Self> { pub(crate) fn build(name: &'static str, threads: u16) -> color_eyre::Result<Self> {
let pool = CpuPool::configure() let pool = CpuPool::configure()
.name(name) .name(name)
.max_threads(threads) .max_threads(threads)