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

Merge branch 'upstream'

Browse Source
This commit is contained in:
Maxime Augier 2023-11-21 18:44:11 +01:00
commit b8728c9440
29 changed files with 1393 additions and 845 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.cargo/config
View File

@ -1,2 +1,2 @@
[build]
# rustflags = ["--cfg", "tokio_unstable"]
rustflags = ["--cfg", "tokio_unstable"]

2
.drone.yml
View File

@ -22,7 +22,7 @@ steps:
pull: always
commands:
- rustup component add clippy
- cargo clippy -- -D warnings
- cargo clippy --no-deps -- -D warnings
trigger:
event:

1129
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

48
Cargo.toml
View File

@ -1,7 +1,7 @@
[package]
name = "ap-relay"
description = "A simple activitypub relay"
version = "0.3.85"
version = "0.3.104"
authors = ["asonix <asonix@asonix.dog>"]
license = "AGPL-3.0"
readme = "README.md"
@ -23,25 +23,20 @@ default = []
[dependencies]
anyhow = "1.0"
actix-rt = "2.7.0"
actix-web = { version = "4.0.1", default-features = false, features = [
"rustls",
"compress-brotli",
"compress-gzip",
] }
actix-webfinger = "0.4.0"
activitystreams = "0.7.0-alpha.21"
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 }
activitystreams = "0.7.0-alpha.25"
activitystreams-ext = "0.1.0-alpha.3"
ammonia = "3.1.0"
awc = { version = "3.0.0", default-features = false, features = ["rustls"] }
bcrypt = "0.14"
bcrypt = "0.15"
base64 = "0.21"
clap = { version = "4.0.0", features = ["derive"] }
config = "0.13.0"
console-subscriber = { version = "0.1", optional = true }
dashmap = "5.1.0"
dotenv = "0.15.0"
futures-util = "0.3.17"
lru = "0.10.0"
flume = "0.11.0"
lru = "0.11.0"
metrics = "0.21.0"
metrics-exporter-prometheus = { version = "0.12.0", default-features = false, features = [
"http-listener",
@ -49,14 +44,18 @@ metrics-exporter-prometheus = { version = "0.12.0", default-features = false, fe
metrics-util = "0.15.0"
mime = "0.3.16"
minify-html = "0.11.0"
opentelemetry = { version = "0.19", features = ["rt-tokio"] }
opentelemetry-otlp = "0.12"
opentelemetry = { version = "0.20", features = ["rt-tokio"] }
opentelemetry-otlp = "0.13"
pin-project-lite = "0.2.9"
quanta = "0.11.0"
rand = "0.8"
rsa = { version = "0.9", features = ["sha2"] }
reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "stream"]}
reqwest-middleware = "0.2"
reqwest-tracing = "0.4.5"
ring = "0.16.20"
rsa = { version = "0.9" }
rsa-magic-public-key = "0.8.0"
rustls = "0.20.7"
rustls = "0.21.0"
rustls-pemfile = "1.0.1"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
@ -69,11 +68,10 @@ teloxide = { version = "0.12.0", default-features = false, features = [
thiserror = "1.0"
time = { version = "0.3.17", features = ["serde"] }
tracing = "0.1"
tracing-awc = "0.1.7"
tracing-error = "0.2"
tracing-futures = "0.2"
tracing-log = "0.1"
tracing-opentelemetry = "0.19"
tracing-opentelemetry = "0.21"
tracing-subscriber = { version = "0.3", features = [
"ansi",
"env-filter",
@ -81,6 +79,7 @@ tracing-subscriber = { version = "0.3", features = [
] }
tokio = { version = "1", features = ["macros", "sync"] }
uuid = { version = "1", features = ["v4", "serde"] }
streem = "0.1.0"
[dependencies.background-jobs]
version = "0.15.0"
@ -88,17 +87,22 @@ default-features = false
features = ["background-jobs-actix", "error-logging"]
[dependencies.http-signature-normalization-actix]
version = "0.8.0"
version = "0.10.1"
default-features = false
features = ["client", "server", "sha-2"]
features = ["server", "ring"]
[dependencies.http-signature-normalization-reqwest]
version = "0.10.0"
default-features = false
features = ["middleware", "ring"]
[dependencies.tracing-actix-web]
version = "0.7.5"
version = "0.7.6"
[build-dependencies]
anyhow = "1.0"
dotenv = "0.15.0"
ructe = { version = "0.16.0", features = ["sass", "mime03"] }
ructe = { version = "0.17.0", features = ["sass", "mime03"] }
toml = "0.7.0"
[profile.dev.package.rsa]

25
README.md
View File

@ -105,7 +105,9 @@ LOCAL_DOMAINS=masto.asonix.dog
LOCAL_BLURB="<p>Welcome to my cool relay where I have cool relay things happening. I hope you enjoy your stay!</p>"
PROMETHEUS_ADDR=0.0.0.0
PROMETHEUS_PORT=9000
CLIENT_POOL_SIZE=20
CLIENT_TIMEOUT=10
DELIVER_CONCURRENCY=8
SIGNATURE_THREADS=2
```
#### Descriptions
@ -155,11 +157,22 @@ Optional - description for the relay
Optional - Address to bind to for serving the prometheus scrape endpoint
##### `PROMETHEUS_PORT`
Optional - Port to bind to for serving the prometheus scrape endpoint
##### `CLIENT_POOL_SIZE`
Optional - How many connections the relay should maintain per thread. This value will be multiplied
by the number of cores available to the relay. This defaults to 20, so a 4-core machine will have a
maximum of 160 simultaneous outbound connections. If you run into problems related to "Too many open
files", you can either decrease this number or increase the ulimit for your system.
##### `CLIENT_TIMEOUT`
Optional - How long the relay will hold open a connection (in seconds) to a remote server during
fetches and deliveries. This defaults to 10
##### `DELIVER_CONCURRENCY`
Optional - How many deliver requests the relay should allow to be in-flight per thread. the default
is 8
##### `SIGNATURE_THREADS`
Optional - Override number of threads used for signing and verifying requests. Default is
`std::thread::available_parallelism()` (It tries to detect how many cores you have). If it cannot
detect the correct number of cores, it falls back to 1.
##### 'PROXY_URL'
Optional - URL of an HTTP proxy to forward outbound requests through
##### 'PROXY_USERNAME'
Optional - username to provide to the HTTP proxy set with `PROXY_URL` through HTTP Basic Auth
##### 'PROXY_PASSWORD'
Optional - password to provide to the HTTP proxy set with `PROXY_URL` through HTTP Basic Auth
### Subscribing
Mastodon admins can subscribe to this relay by adding the `/inbox` route to their relay settings.

12
flake.lock
View File

@ -5,11 +5,11 @@
"systems": "systems"
},
"locked": {
"lastModified": 1687171271,
"narHash": "sha256-BJlq+ozK2B1sJDQXS3tzJM5a+oVZmi1q0FlBK/Xqv7M=",
"lastModified": 1692799911,
"narHash": "sha256-3eihraek4qL744EvQXsK1Ha6C3CR7nnT8X2qWap4RNk=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "abfb11bd1aec8ced1c9bb9adfe68018230f4fb3c",
"rev": "f9e7cf818399d17d347f847525c5a5a8032e4e44",
"type": "github"
},
"original": {
@ -20,11 +20,11 @@
},
"nixpkgs": {
"locked": {
"lastModified": 1687412861,
"narHash": "sha256-Z/g0wbL68C+mSGerYS2quv9FXQ1RRP082cAC0Bh4vcs=",
"lastModified": 1693003285,
"narHash": "sha256-5nm4yrEHKupjn62MibENtfqlP6pWcRTuSKrMiH9bLkc=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "e603dc5f061ca1d8a19b3ede6a8cf9c9fcba6cdc",
"rev": "5690c4271f2998c304a45c91a0aeb8fb69feaea7",
"type": "github"
},
"original": {

5
relay.nix
View File

@ -6,19 +6,20 @@
rustPlatform.buildRustPackage {
pname = "relay";
version = "0.3.85";
version = "0.3.104";
src = ./.;
cargoLock.lockFile = ./Cargo.lock;
PROTOC = "${protobuf}/bin/protoc";
PROTOC_INCLUDE = "${protobuf}/include";
RUSTFLAGS = "--cfg tokio_unstable";
nativeBuildInputs = [ ];
passthru.tests = { inherit (nixosTests) relay; };
meta = with lib; {
description = "A simple image hosting service";
description = "An ActivityPub relay";
homepage = "https://git.asonix.dog/asonix/relay";
license = with licenses; [ agpl3Plus ];
};

50
src/admin/client.rs
View File

@ -3,12 +3,14 @@ use crate::{
collector::Snapshot,
config::{AdminUrlKind, Config},
error::{Error, ErrorKind},
extractors::XApiToken,
};
use awc::Client;
use actix_web::http::header::Header;
use reqwest_middleware::ClientWithMiddleware;
use serde::de::DeserializeOwned;
pub(crate) async fn allow(
client: &Client,
client: &ClientWithMiddleware,
config: &Config,
domains: Vec<String>,
) -> Result<(), Error> {
@ -16,7 +18,7 @@ pub(crate) async fn allow(
}
pub(crate) async fn disallow(
client: &Client,
client: &ClientWithMiddleware,
config: &Config,
domains: Vec<String>,
) -> Result<(), Error> {
@ -24,7 +26,7 @@ pub(crate) async fn disallow(
}
pub(crate) async fn block(
client: &Client,
client: &ClientWithMiddleware,
config: &Config,
domains: Vec<String>,
) -> Result<(), Error> {
@ -32,35 +34,50 @@ pub(crate) async fn block(
}
pub(crate) async fn unblock(
client: &Client,
client: &ClientWithMiddleware,
config: &Config,
domains: Vec<String>,
) -> Result<(), Error> {
post_domains(client, config, domains, AdminUrlKind::Unblock).await
}
pub(crate) async fn allowed(client: &Client, config: &Config) -> Result<AllowedDomains, Error> {
pub(crate) async fn allowed(
client: &ClientWithMiddleware,
config: &Config,
) -> Result<AllowedDomains, Error> {
get_results(client, config, AdminUrlKind::Allowed).await
}
pub(crate) async fn blocked(client: &Client, config: &Config) -> Result<BlockedDomains, Error> {
pub(crate) async fn blocked(
client: &ClientWithMiddleware,
config: &Config,
) -> Result<BlockedDomains, Error> {
get_results(client, config, AdminUrlKind::Blocked).await
}
pub(crate) async fn connected(client: &Client, config: &Config) -> Result<ConnectedActors, Error> {
pub(crate) async fn connected(
client: &ClientWithMiddleware,
config: &Config,
) -> Result<ConnectedActors, Error> {
get_results(client, config, AdminUrlKind::Connected).await
}
pub(crate) async fn stats(client: &Client, config: &Config) -> Result<Snapshot, Error> {
pub(crate) async fn stats(
client: &ClientWithMiddleware,
config: &Config,
) -> Result<Snapshot, Error> {
get_results(client, config, AdminUrlKind::Stats).await
}
pub(crate) async fn last_seen(client: &Client, config: &Config) -> Result<LastSeen, Error> {
pub(crate) async fn last_seen(
client: &ClientWithMiddleware,
config: &Config,
) -> Result<LastSeen, Error> {
get_results(client, config, AdminUrlKind::LastSeen).await
}
async fn get_results<T: DeserializeOwned>(
client: &Client,
client: &ClientWithMiddleware,
config: &Config,
url_kind: AdminUrlKind,
) -> Result<T, Error> {
@ -68,9 +85,9 @@ async fn get_results<T: DeserializeOwned>(
let iri = config.generate_admin_url(url_kind);
let mut res = client
let res = client
.get(iri.as_str())
.insert_header(x_api_token)
.header(XApiToken::name(), x_api_token.to_string())
.send()
.await
.map_err(|e| ErrorKind::SendRequest(iri.to_string(), e.to_string()))?;
@ -88,7 +105,7 @@ async fn get_results<T: DeserializeOwned>(
}
async fn post_domains(
client: &Client,
client: &ClientWithMiddleware,
config: &Config,
domains: Vec<String>,
url_kind: AdminUrlKind,
@ -99,8 +116,9 @@ async fn post_domains(
let res = client
.post(iri.as_str())
.insert_header(x_api_token)
.send_json(&Domains { domains })
.header(XApiToken::name(), x_api_token.to_string())
.json(&Domains { domains })
.send()
.await
.map_err(|e| ErrorKind::SendRequest(iri.to_string(), e.to_string()))?;

2
src/collector.rs
View File

@ -159,7 +159,7 @@ impl Snapshot {
let entry = merging.entry(name).or_insert_with(HashMap::new);
for counter in counters {
let mut merge_counter = entry
let merge_counter = entry
.entry(counter.labels.clone())
.or_insert_with(MergeCounter::default);
if key == *start {

103
src/config.rs
View File

@ -11,8 +11,7 @@ use activitystreams::{
},
};
use config::Environment;
use http_signature_normalization_actix::prelude::VerifyDigest;
use rsa::sha2::{Digest, Sha256};
use http_signature_normalization_actix::{digest::ring::Sha256, prelude::VerifyDigest};
use rustls::{Certificate, PrivateKey};
use std::{
io::BufReader,
@ -45,7 +44,12 @@ pub(crate) struct ParsedConfig {
local_blurb: Option<String>,
prometheus_addr: Option<IpAddr>,
prometheus_port: Option<u16>,
client_pool_size: usize,
deliver_concurrency: u64,
client_timeout: u64,
proxy_url: Option<IriString>,
proxy_username: Option<String>,
proxy_password: Option<String>,
signature_threads: Option<usize>,
}
#[derive(Clone)]
@ -69,7 +73,10 @@ pub struct Config {
local_domains: Vec<String>,
local_blurb: Option<String>,
prometheus_config: Option<PrometheusConfig>,
client_pool_size: usize,
deliver_concurrency: u64,
client_timeout: u64,
proxy_config: Option<ProxyConfig>,
signature_threads: Option<usize>,
}
#[derive(Clone)]
@ -84,6 +91,12 @@ struct PrometheusConfig {
port: u16,
}
#[derive(Clone, Debug)]
struct ProxyConfig {
url: IriString,
auth: Option<(String, String)>,
}
#[derive(Debug)]
pub enum UrlKind {
Activity,
@ -137,7 +150,10 @@ impl std::fmt::Debug for Config {
.field("local_domains", &self.local_domains)
.field("local_blurb", &self.local_blurb)
.field("prometheus_config", &self.prometheus_config)
.field("client_pool_size", &self.client_pool_size)
.field("deliver_concurrency", &self.deliver_concurrency)
.field("client_timeout", &self.client_timeout)
.field("proxy_config", &self.proxy_config)
.field("signature_threads", &self.signature_threads)
.finish()
}
}
@ -167,7 +183,12 @@ impl Config {
.set_default("local_blurb", None as Option<&str>)?
.set_default("prometheus_addr", None as Option<&str>)?
.set_default("prometheus_port", None as Option<u16>)?
.set_default("client_pool_size", 20u64)?
.set_default("deliver_concurrency", 8u64)?
.set_default("client_timeout", 10u64)?
.set_default("proxy_url", None as Option<&str>)?
.set_default("proxy_username", None as Option<&str>)?
.set_default("proxy_password", None as Option<&str>)?
.set_default("signature_threads", None as Option<u64>)?
.add_source(Environment::default())
.build()?;
@ -209,6 +230,26 @@ impl Config {
(None, None) => None,
};
let proxy_config = match (config.proxy_username, config.proxy_password) {
(Some(username), Some(password)) => config.proxy_url.map(|url| ProxyConfig {
url,
auth: Some((username, password)),
}),
(Some(_), None) => {
tracing::warn!(
"PROXY_USERNAME is set but PROXY_PASSWORD is not set, not setting Proxy Auth"
);
config.proxy_url.map(|url| ProxyConfig { url, auth: None })
}
(None, Some(_)) => {
tracing::warn!(
"PROXY_PASSWORD is set but PROXY_USERNAME is not set, not setting Proxy Auth"
);
config.proxy_url.map(|url| ProxyConfig { url, auth: None })
}
(None, None) => config.proxy_url.map(|url| ProxyConfig { url, auth: None }),
};
let source_url = match Self::git_hash() {
Some(hash) => format!(
"{}{}{hash}",
@ -239,10 +280,32 @@ impl Config {
local_domains,
local_blurb: config.local_blurb,
prometheus_config,
client_pool_size: config.client_pool_size,
deliver_concurrency: config.deliver_concurrency,
client_timeout: config.client_timeout,
proxy_config,
signature_threads: config.signature_threads,
})
}
pub(crate) fn signature_threads(&self) -> usize {
self.signature_threads
.unwrap_or_else(|| {
std::thread::available_parallelism()
.map(usize::from)
.map_err(|e| tracing::warn!("Failed to get parallelism, {e}"))
.unwrap_or(1)
})
.max(1)
}
pub(crate) fn client_timeout(&self) -> u64 {
self.client_timeout
}
pub(crate) fn deliver_concurrency(&self) -> u64 {
self.deliver_concurrency
}
pub(crate) fn prometheus_bind_address(&self) -> Option<SocketAddr> {
let config = self.prometheus_config.as_ref()?;
@ -291,7 +354,15 @@ impl Config {
pub(crate) fn footer_blurb(&self) -> Option<crate::templates::Html<String>> {
if let Some(blurb) = &self.footer_blurb {
if !blurb.is_empty() {
return Some(crate::templates::Html(ammonia::clean(blurb)));
return Some(crate::templates::Html(
ammonia::Builder::new()
.add_tag_attributes("a", &["rel"])
.add_tag_attributes("area", &["rel"])
.add_tag_attributes("link", &["rel"])
.link_rel(None)
.clean(blurb)
.to_string(),
));
}
}
@ -301,7 +372,15 @@ impl Config {
pub(crate) fn local_blurb(&self) -> Option<crate::templates::Html<String>> {
if let Some(blurb) = &self.local_blurb {
if !blurb.is_empty() {
return Some(crate::templates::Html(ammonia::clean(blurb)));
return Some(crate::templates::Html(
ammonia::Builder::new()
.add_tag_attributes("a", &["rel"])
.add_tag_attributes("area", &["rel"])
.add_tag_attributes("link", &["rel"])
.link_rel(None)
.clean(blurb)
.to_string(),
));
}
}
@ -419,8 +498,10 @@ impl Config {
)
}
pub(crate) fn client_pool_size(&self) -> usize {
self.client_pool_size
pub(crate) fn proxy_config(&self) -> Option<(&IriString, Option<(&str, &str)>)> {
self.proxy_config.as_ref().map(|ProxyConfig { url, auth }| {
(url, auth.as_ref().map(|(u, p)| (u.as_str(), p.as_str())))
})
}
pub(crate) fn source_code(&self) -> &IriString {

8
src/data/actor.rs
View File

@ -2,7 +2,7 @@ use crate::{
apub::AcceptedActors,
db::{Actor, Db},
error::{Error, ErrorKind},
requests::Requests,
requests::{BreakerStrategy, Requests},
};
use activitystreams::{iri_string::types::IriString, prelude::*};
use std::time::{Duration, SystemTime};
@ -71,7 +71,9 @@ impl ActorCache {
id: &IriString,
requests: &Requests,
) -> Result<Actor, Error> {
let accepted_actor = requests.fetch::<AcceptedActors>(id).await?;
let accepted_actor = requests
.fetch::<AcceptedActors>(id, BreakerStrategy::Require2XX)
.await?;
let input_authority = id.authority_components().ok_or(ErrorKind::MissingDomain)?;
let accepted_actor_id = accepted_actor
@ -97,6 +99,6 @@ impl ActorCache {
fn get_inbox(actor: &AcceptedActors) -> Result<&IriString, Error> {
Ok(actor
.endpoints()?
.and_then(|e| e.shared_inbox)
.and_then(|e| e.shared_inbox.as_ref())
.unwrap_or(actor.inbox()?))
}

53
src/data/state.rs
View File

@ -1,14 +1,15 @@
use crate::{
config::{Config, UrlKind},
data::NodeCache,
db::Db,
error::Error,
requests::{Breakers, Requests},
spawner::Spawner,
};
use activitystreams::iri_string::types::IriString;
use actix_web::web;
use lru::LruCache;
use rand::thread_rng;
use reqwest_middleware::ClientWithMiddleware;
use rsa::{RsaPrivateKey, RsaPublicKey};
use std::sync::{Arc, RwLock};
@ -16,10 +17,10 @@ use super::LastOnline;
#[derive(Clone)]
pub struct State {
pub(crate) requests: Requests,
pub(crate) public_key: RsaPublicKey,
private_key: RsaPrivateKey,
object_cache: Arc<RwLock<LruCache<IriString, IriString>>>,
node_cache: NodeCache,
pub(crate) node_cache: NodeCache,
breakers: Breakers,
pub(crate) last_online: Arc<LastOnline>,
pub(crate) db: Db,
@ -36,21 +37,6 @@ impl std::fmt::Debug for State {
}
impl State {
pub(crate) fn node_cache(&self) -> NodeCache {
self.node_cache.clone()
}
pub(crate) fn requests(&self, config: &Config) -> Requests {
Requests::new(
config.generate_url(UrlKind::MainKey).to_string(),
self.private_key.clone(),
config.user_agent(),
self.breakers.clone(),
self.last_online.clone(),
config.client_pool_size(),
)
}
#[tracing::instrument(
level = "debug",
name = "Get inboxes for other domains",
@ -90,8 +76,17 @@ impl State {
self.object_cache.write().unwrap().put(object_id, actor_id);
}
pub(crate) fn is_connected(&self, iri: &IriString) -> bool {
self.breakers.should_try(iri)
}
#[tracing::instrument(level = "debug", name = "Building state", skip_all)]
pub(crate) async fn build(db: Db) -> Result<Self, Error> {
pub(crate) async fn build(
db: Db,
key_id: String,
spawner: Spawner,
client: ClientWithMiddleware,
) -> Result<Self, Error> {
let private_key = if let Ok(Some(key)) = db.private_key().await {
tracing::debug!("Using existing key");
key
@ -110,16 +105,28 @@ impl State {
let public_key = private_key.to_public_key();
let state = State {
public_key,
let breakers = Breakers::default();
let last_online = Arc::new(LastOnline::empty());
let requests = Requests::new(
key_id,
private_key,
breakers.clone(),
last_online.clone(),
spawner,
client,
);
let state = State {
requests,
public_key,
object_cache: Arc::new(RwLock::new(LruCache::new(
(1024 * 8).try_into().expect("nonzero"),
))),
node_cache: NodeCache::new(db.clone()),
breakers: Breakers::default(),
breakers,
db,
last_online: Arc::new(LastOnline::empty()),
last_online,
};
Ok(state)

35
src/error.rs
View File

@ -5,7 +5,7 @@ use actix_web::{
http::StatusCode,
HttpResponse,
};
use http_signature_normalization_actix::PrepareSignError;
use http_signature_normalization_reqwest::SignError;
use std::{convert::Infallible, fmt::Debug, io};
use tracing_error::SpanTrace;
@ -81,6 +81,15 @@ pub(crate) enum ErrorKind {
#[error("Couldn't encode public key, {0}")]
Spki(#[from] rsa::pkcs8::spki::Error),
#[error("Couldn't sign request")]
SignRequest,
#[error("Couldn't make request")]
Reqwest(#[from] reqwest::Error),
#[error("Couldn't build client")]
ReqwestMiddleware(#[from] reqwest_middleware::Error),
#[error("Couldn't parse IRI, {0}")]
ParseIri(#[from] activitystreams::iri_string::validate::Error),
@ -99,17 +108,17 @@ pub(crate) enum ErrorKind {
#[error("Couldn't do the json thing, {0}")]
Json(#[from] serde_json::Error),
#[error("Couldn't build signing string, {0}")]
PrepareSign(#[from] PrepareSignError),
#[error("Couldn't sign request, {0}")]
Sign(#[from] SignError),
#[error("Couldn't sign digest")]
Signature(#[from] rsa::signature::Error),
#[error("Couldn't read signature")]
ReadSignature(rsa::signature::Error),
#[error("Couldn't verify signature")]
VerifySignature(rsa::signature::Error),
VerifySignature,
#[error("Failed to encode key der")]
DerEncode,
#[error("Couldn't parse the signature header")]
HeaderValidation(#[from] actix_web::http::header::InvalidHeaderValue),
@ -242,3 +251,15 @@ impl From<rsa::errors::Error> for ErrorKind {
ErrorKind::Rsa(e)
}
}
impl From<http_signature_normalization_actix::Canceled> for ErrorKind {
fn from(_: http_signature_normalization_actix::Canceled) -> Self {
Self::Canceled
}
}
impl From<http_signature_normalization_reqwest::Canceled> for ErrorKind {
fn from(_: http_signature_normalization_reqwest::Canceled) -> Self {
Self::Canceled
}
}

46
src/extractors.rs
View File

@ -1,6 +1,6 @@
use actix_web::{
dev::Payload,
error::{BlockingError, ParseError},
error::ParseError,
http::{
header::{from_one_raw_str, Header, HeaderName, HeaderValue, TryIntoHeaderValue},
StatusCode,
@ -9,12 +9,11 @@ use actix_web::{
FromRequest, HttpMessage, HttpRequest, HttpResponse, ResponseError,
};
use bcrypt::{BcryptError, DEFAULT_COST};
use futures_util::future::LocalBoxFuture;
use http_signature_normalization_actix::prelude::InvalidHeaderValue;
use http_signature_normalization_actix::{prelude::InvalidHeaderValue, Canceled, Spawn};
use std::{convert::Infallible, str::FromStr, time::Instant};
use tracing_error::SpanTrace;
use crate::db::Db;
use crate::{db::Db, future::LocalBoxFuture, spawner::Spawner};
#[derive(Clone)]
pub(crate) struct AdminConfig {
@ -37,10 +36,10 @@ pub(crate) struct Admin {
db: Data<Db>,
}
type PrepareTuple = (Data<Db>, Data<AdminConfig>, Data<Spawner>, XApiToken);
impl Admin {
fn prepare_verify(
req: &HttpRequest,
) -> Result<(Data<Db>, Data<AdminConfig>, XApiToken), Error> {
fn prepare_verify(req: &HttpRequest) -> Result<PrepareTuple, Error> {
let hashed_api_token = req
.app_data::<Data<AdminConfig>>()
.ok_or_else(Error::missing_config)?
@ -53,16 +52,23 @@ impl Admin {
.ok_or_else(Error::missing_db)?
.clone();
Ok((db, hashed_api_token, x_api_token))
let spawner = req
.app_data::<Data<Spawner>>()
.ok_or_else(Error::missing_spawner)?
.clone();
Ok((db, hashed_api_token, spawner, x_api_token))
}
#[tracing::instrument(level = "debug", skip_all)]
async fn verify(
hashed_api_token: Data<AdminConfig>,
spawner: Data<Spawner>,
x_api_token: XApiToken,
) -> Result<(), Error> {
let span = tracing::Span::current();
if actix_web::web::block(move || span.in_scope(|| hashed_api_token.verify(x_api_token)))
if spawner
.spawn_blocking(move || span.in_scope(|| hashed_api_token.verify(x_api_token)))
.await
.map_err(Error::canceled)??
{
@ -107,6 +113,13 @@ impl Error {
}
}
fn missing_spawner() -> Self {
Error {
context: SpanTrace::capture().to_string(),
kind: ErrorKind::MissingSpawner,
}
}
fn bcrypt_verify(e: BcryptError) -> Self {
Error {
context: SpanTrace::capture().to_string(),
@ -128,7 +141,7 @@ impl Error {
}
}
fn canceled(_: BlockingError) -> Self {
fn canceled(_: Canceled) -> Self {
Error {
context: SpanTrace::capture().to_string(),
kind: ErrorKind::Canceled,
@ -147,6 +160,9 @@ enum ErrorKind {
#[error("Missing Db")]
MissingDb,
#[error("Missing Spawner")]
MissingSpawner,
#[error("Panic in verify")]
Canceled,
@ -182,8 +198,8 @@ impl FromRequest for Admin {
let now = Instant::now();
let res = Self::prepare_verify(req);
Box::pin(async move {
let (db, c, t) = res?;
Self::verify(c, t).await?;
let (db, c, s, t) = res?;
Self::verify(c, s, t).await?;
metrics::histogram!(
"relay.admin.verify",
now.elapsed().as_micros() as f64 / 1_000_000_f64
@ -226,3 +242,9 @@ impl FromStr for XApiToken {
Ok(XApiToken(s.to_string()))
}
}
impl std::fmt::Display for XApiToken {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.0.fmt(f)
}
}

3
src/future.rs Normal file
View File

@ -0,0 +1,3 @@
use std::{future::Future, pin::Pin};
pub(crate) type LocalBoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + 'a>>;

13
src/jobs.rs
View File

@ -14,10 +14,9 @@ pub(crate) use self::{
use crate::{
config::Config,
data::{ActorCache, MediaCache, NodeCache, State},
data::{ActorCache, MediaCache, State},
error::{Error, ErrorKind},
jobs::{process_listeners::Listeners, record_last_online::RecordLastOnline},
requests::Requests,
};
use background_jobs::{
memory_storage::{ActixTimer, Storage},
@ -45,6 +44,8 @@ pub(crate) fn create_workers(
media: MediaCache,
config: Config,
) -> JobServer {
let deliver_concurrency = config.deliver_concurrency();
let queue_handle = WorkerConfig::new(Storage::new(ActixTimer), move |queue_handle| {
JobState::new(
state.clone(),
@ -68,7 +69,7 @@ pub(crate) fn create_workers(
.register::<apub::Undo>()
.set_worker_count("maintenance", 2)
.set_worker_count("apub", 2)
.set_worker_count("deliver", 8)
.set_worker_count("deliver", deliver_concurrency)
.start();
queue_handle.every(Duration::from_secs(60 * 5), Listeners);
@ -79,12 +80,10 @@ pub(crate) fn create_workers(
#[derive(Clone, Debug)]
pub(crate) struct JobState {
requests: Requests,
state: State,
actors: ActorCache,
config: Config,
media: MediaCache,
node_cache: NodeCache,
job_server: JobServer,
}
@ -110,12 +109,10 @@ impl JobState {
config: Config,
) -> Self {
JobState {
requests: state.requests(&config),
node_cache: state.node_cache(),
state,
actors,
config,
media,
state,
job_server,
}
}

6
src/jobs/contact.rs
View File

@ -2,6 +2,7 @@ use crate::{
apub::AcceptedActors,
error::{Error, ErrorKind},
jobs::JobState,
requests::BreakerStrategy,
};
use activitystreams::{iri_string::types::IriString, object::Image, prelude::*};
use background_jobs::ActixJob;
@ -32,6 +33,7 @@ impl QueryContact {
async fn perform(self, state: JobState) -> Result<(), Error> {
let contact_outdated = state
.state
.node_cache
.is_contact_outdated(self.actor_id.clone())
.await;
@ -41,8 +43,9 @@ impl QueryContact {
}
let contact = match state
.state
.requests
.fetch::<AcceptedActors>(&self.contact_id)
.fetch::<AcceptedActors>(&self.contact_id, BreakerStrategy::Allow404AndBelow)
.await
{
Ok(contact) => contact,
@ -57,6 +60,7 @@ impl QueryContact {
to_contact(contact).ok_or(ErrorKind::Extract("contact"))?;
state
.state
.node_cache
.set_contact(self.actor_id, username, display_name, url, avatar)
.await?;

8
src/jobs/deliver.rs
View File

@ -1,6 +1,7 @@
use crate::{
error::Error,
jobs::{debug_object, JobState},
requests::BreakerStrategy,
};
use activitystreams::iri_string::types::IriString;
use background_jobs::{ActixJob, Backoff};
@ -35,7 +36,12 @@ impl Deliver {
#[tracing::instrument(name = "Deliver", skip(state))]
async fn permform(self, state: JobState) -> Result<(), Error> {
if let Err(e) = state.requests.deliver(&self.to, &self.data).await {
if let Err(e) = state
.state
.requests
.deliver(&self.to, &self.data, BreakerStrategy::Allow401AndBelow)
.await
{
if e.is_breaker() {
tracing::debug!("Not trying due to failed breaker");
return Ok(());

2
src/jobs/deliver_many.rs
View File

@ -1,10 +1,10 @@
use crate::{
error::Error,
future::LocalBoxFuture,
jobs::{debug_object, Deliver, JobState},
};
use activitystreams::iri_string::types::IriString;
use background_jobs::ActixJob;
use futures_util::future::LocalBoxFuture;
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub(crate) struct DeliverMany {

17
src/jobs/instance.rs
View File

@ -2,6 +2,7 @@ use crate::{
config::UrlKind,
error::{Error, ErrorKind},
jobs::{Boolish, JobState},
requests::BreakerStrategy,
};
use activitystreams::{iri, iri_string::types::IriString};
use background_jobs::ActixJob;
@ -40,15 +41,23 @@ impl QueryInstance {
InstanceApiType::Mastodon => {
let mastodon_instance_uri = iri!(format!("{scheme}://{authority}/api/v1/instance"));
state
.state
.requests
.fetch_json::<Instance>(&mastodon_instance_uri)
.fetch_json::<Instance>(
&mastodon_instance_uri,
BreakerStrategy::Allow404AndBelow,
)
.await
}
InstanceApiType::Misskey => {
let msky_meta_uri = iri!(format!("{scheme}://{authority}/api/meta"));
state
.state
.requests
.fetch_json_msky::<MisskeyMeta>(&msky_meta_uri)
.fetch_json_msky::<MisskeyMeta>(
&msky_meta_uri,
BreakerStrategy::Allow404AndBelow,
)
.await
.map(|res| res.into())
}
@ -58,10 +67,12 @@ impl QueryInstance {
#[tracing::instrument(name = "Query instance", skip(state))]
async fn perform(self, state: JobState) -> Result<(), Error> {
let contact_outdated = state
.state
.node_cache
.is_contact_outdated(self.actor_id.clone())
.await;
let instance_outdated = state
.state
.node_cache
.is_instance_outdated(self.actor_id.clone())
.await;
@ -123,6 +134,7 @@ impl QueryInstance {
let avatar = state.config.generate_url(UrlKind::Media(uuid));
state
.state
.node_cache
.set_contact(
self.actor_id.clone(),
@ -137,6 +149,7 @@ impl QueryInstance {
let description = ammonia::clean(&description);
state
.state
.node_cache
.set_instance(
self.actor_id,

13
src/jobs/nodeinfo.rs
View File

@ -1,6 +1,7 @@
use crate::{
error::{Error, ErrorKind},
jobs::{Boolish, JobState, QueryContact},
requests::BreakerStrategy,
};
use activitystreams::{iri, iri_string::types::IriString, primitives::OneOrMany};
use background_jobs::ActixJob;
@ -27,6 +28,7 @@ impl QueryNodeinfo {
#[tracing::instrument(name = "Query node info", skip(state))]
async fn perform(self, state: JobState) -> Result<(), Error> {
if !state
.state
.node_cache
.is_nodeinfo_outdated(self.actor_id.clone())
.await
@ -42,8 +44,9 @@ impl QueryNodeinfo {
let well_known_uri = iri!(format!("{scheme}://{authority}/.well-known/nodeinfo"));
let well_known = match state
.state
.requests
.fetch_json::<WellKnown>(&well_known_uri)
.fetch_json::<WellKnown>(&well_known_uri, BreakerStrategy::Allow404AndBelow)
.await
{
Ok(well_known) => well_known,
@ -60,7 +63,12 @@ impl QueryNodeinfo {
return Ok(());
};
let nodeinfo = match state.requests.fetch_json::<Nodeinfo>(&href).await {
let nodeinfo = match state
.state
.requests
.fetch_json::<Nodeinfo>(&href, BreakerStrategy::Require2XX)
.await
{
Ok(nodeinfo) => nodeinfo,
Err(e) if e.is_breaker() => {
tracing::debug!("Not retrying due to failed breaker");
@ -70,6 +78,7 @@ impl QueryNodeinfo {
};
state
.state
.node_cache
.set_info(
self.actor_id.clone(),

95
src/main.rs
View File

@ -1,17 +1,21 @@
// need this for ructe
#![allow(clippy::needless_borrow)]
use std::time::Duration;
use activitystreams::iri_string::types::IriString;
use actix_rt::task::JoinHandle;
use actix_web::{middleware::Compress, web, App, HttpServer};
use collector::MemoryCollector;
#[cfg(feature = "console")]
use console_subscriber::ConsoleLayer;
use error::Error;
use http_signature_normalization_actix::middleware::VerifySignature;
use metrics_exporter_prometheus::PrometheusBuilder;
use metrics_util::layers::FanoutBuilder;
use opentelemetry::{sdk::Resource, KeyValue};
use opentelemetry_otlp::WithExportConfig;
use reqwest_middleware::ClientWithMiddleware;
use rustls::ServerConfig;
use tracing_actix_web::TracingLogger;
use tracing_error::ErrorLayer;
@ -27,12 +31,16 @@ mod data;
mod db;
mod error;
mod extractors;
mod future;
mod jobs;
mod middleware;
mod requests;
mod routes;
mod spawner;
mod telegram;
use crate::config::UrlKind;
use self::{
args::Args,
config::Config,
@ -41,6 +49,7 @@ use self::{
jobs::create_workers,
middleware::{DebugPayload, MyVerify, RelayResolver, Timings},
routes::{actor, healthz, inbox, index, nodeinfo, nodeinfo_meta, statics},
spawner::Spawner,
};
fn init_subscriber(
@ -98,6 +107,38 @@ fn init_subscriber(
Ok(())
}
fn build_client(
user_agent: &str,
timeout_seconds: u64,
proxy: Option<(&IriString, Option<(&str, &str)>)>,
) -> Result<ClientWithMiddleware, Error> {
let builder = reqwest::Client::builder().user_agent(user_agent.to_string());
let builder = if let Some((url, auth)) = proxy {
let proxy = reqwest::Proxy::all(url.as_str())?;
let proxy = if let Some((username, password)) = auth {
proxy.basic_auth(username, password)
} else {
proxy
};
builder.proxy(proxy)
} else {
builder
};
let client = builder
.timeout(Duration::from_secs(timeout_seconds))
.build()?;
let client_with_middleware = reqwest_middleware::ClientBuilder::new(client)
.with(reqwest_tracing::TracingMiddleware::default())
.build();
Ok(client_with_middleware)
}
#[actix_rt::main]
async fn main() -> Result<(), anyhow::Error> {
dotenv::dotenv().ok();
@ -148,7 +189,11 @@ fn client_main(config: Config, args: Args) -> JoinHandle<Result<(), anyhow::Erro
}
async fn do_client_main(config: Config, args: Args) -> Result<(), anyhow::Error> {
let client = requests::build_client(&config.user_agent(), config.client_pool_size());
let client = build_client(
&config.user_agent(),
config.client_timeout(),
config.proxy_config(),
)?;
if !args.blocks().is_empty() || !args.allowed().is_empty() {
if args.undo() {
@ -236,6 +281,8 @@ fn server_main(
actix_rt::spawn(do_server_main(db, actors, media, collector, config))
}
const VERIFY_RATIO: usize = 7;
async fn do_server_main(
db: Db,
actors: ActorCache,
@ -243,8 +290,30 @@ async fn do_server_main(
collector: MemoryCollector,
config: Config,
) -> Result<(), anyhow::Error> {
let client = build_client(
&config.user_agent(),
config.client_timeout(),
config.proxy_config(),
)?;
tracing::warn!("Creating state");
let state = State::build(db.clone()).await?;
let (signature_threads, verify_threads) = match config.signature_threads() {
0 | 1 => (1, 1),
n if n <= VERIFY_RATIO => (n, 1),
n => {
let verify_threads = (n / VERIFY_RATIO).max(1);
let signature_threads = n.saturating_sub(verify_threads).max(VERIFY_RATIO);
(signature_threads, verify_threads)
}
};
let verify_spawner = Spawner::build("verify-cpu", verify_threads)?;
let sign_spawner = Spawner::build("sign-cpu", signature_threads)?;
let key_id = config.generate_url(UrlKind::MainKey).to_string();
let state = State::build(db.clone(), key_id, sign_spawner, client).await?;
if let Some((token, admin_handle)) = config.telegram_info() {
tracing::warn!("Creating telegram handler");
@ -255,20 +324,21 @@ async fn do_server_main(
let bind_address = config.bind_address();
let server = HttpServer::new(move || {
let requests = state.requests(&config);
let job_server =
create_workers(state.clone(), actors.clone(), media.clone(), config.clone());
let app = App::new()
.app_data(web::Data::new(db.clone()))
.app_data(web::Data::new(state.clone()))
.app_data(web::Data::new(requests.clone()))
.app_data(web::Data::new(
state.requests.clone().spawner(verify_spawner.clone()),
))
.app_data(web::Data::new(actors.clone()))
.app_data(web::Data::new(config.clone()))
.app_data(web::Data::new(job_server))
.app_data(web::Data::new(media.clone()))
.app_data(web::Data::new(collector.clone()));
.app_data(web::Data::new(collector.clone()))
.app_data(web::Data::new(verify_spawner.clone()));
let app = if let Some(data) = config.admin_config() {
app.app_data(data)
@ -284,10 +354,15 @@ async fn do_server_main(
.service(web::resource("/media/{path}").route(web::get().to(routes::media)))
.service(
web::resource("/inbox")
.wrap(config.digest_middleware())
.wrap(config.digest_middleware().spawner(verify_spawner.clone()))
.wrap(VerifySignature::new(
MyVerify(requests, actors.clone(), state.clone()),
Default::default(),
MyVerify(
state.requests.clone().spawner(verify_spawner.clone()),
actors.clone(),
state.clone(),
verify_spawner.clone(),
),
http_signature_normalization_actix::Config::new(),
))
.wrap(DebugPayload(config.debug()))
.route(web::post().to(inbox)),
@ -325,7 +400,7 @@ async fn do_server_main(
.with_no_client_auth()
.with_single_cert(certs, key)?;
server
.bind_rustls(bind_address, server_config)?
.bind_rustls_021(bind_address, server_config)?
.run()
.await?;
} else {

31
src/middleware/payload.rs
View File

@ -4,14 +4,11 @@ use actix_web::{
web::BytesMut,
HttpMessage,
};
use futures_util::{
future::TryFutureExt,
stream::{once, TryStreamExt},
};
use std::{
future::{ready, Ready},
task::{Context, Poll},
};
use streem::IntoStreamer;
#[derive(Clone, Debug)]
pub(crate) struct DebugPayload(pub bool);
@ -53,19 +50,23 @@ where
fn call(&self, mut req: ServiceRequest) -> Self::Future {
if self.0 && req.method() == Method::POST {
let pl = req.take_payload();
let mut pl = req.take_payload().into_streamer();
req.set_payload(Payload::Stream {
payload: Box::pin(once(
pl.try_fold(BytesMut::new(), |mut acc, bytes| async {
acc.extend(bytes);
Ok(acc)
})
.map_ok(|bytes| {
let bytes = bytes.freeze();
payload: Box::pin(streem::try_from_fn(|yielder| async move {
let mut buf = BytesMut::new();
while let Some(bytes) = pl.try_next().await? {
buf.extend(bytes);
}
let bytes = buf.freeze();
tracing::info!("{}", String::from_utf8_lossy(&bytes));
bytes
}),
)),
yielder.yield_ok(bytes).await;
Ok(())
})),
});
self.1.call(req)

49
src/middleware/verifier.rs
View File

@ -2,20 +2,17 @@ use crate::{
apub::AcceptedActors,
data::{ActorCache, State},
error::{Error, ErrorKind},
requests::Requests,
requests::{BreakerStrategy, Requests},
spawner::Spawner,
};
use activitystreams::{base::BaseExt, iri, iri_string::types::IriString};
use actix_web::web;
use base64::{engine::general_purpose::STANDARD, Engine};
use http_signature_normalization_actix::{prelude::*, verify::DeprecatedAlgorithm};
use rsa::{
pkcs1v15::Signature, pkcs1v15::VerifyingKey, pkcs8::DecodePublicKey, sha2::Sha256,
signature::Verifier, RsaPublicKey,
};
use http_signature_normalization_actix::{prelude::*, verify::DeprecatedAlgorithm, Spawn};
use rsa::{pkcs1::EncodeRsaPublicKey, pkcs8::DecodePublicKey, RsaPublicKey};
use std::{future::Future, pin::Pin};
#[derive(Clone, Debug)]
pub(crate) struct MyVerify(pub Requests, pub ActorCache, pub State);
pub(crate) struct MyVerify(pub Requests, pub ActorCache, pub State, pub Spawner);
impl MyVerify {
#[tracing::instrument("Verify request", skip(self, signature, signing_string))]
@ -55,7 +52,13 @@ impl MyVerify {
None => (),
};
let res = do_verify(&actor.public_key, signature.clone(), signing_string.clone()).await;
let res = do_verify(
&self.3,
&actor.public_key,
signature.clone(),
signing_string.clone(),
)
.await;
if let Err(e) = res {
if !was_cached {
@ -67,7 +70,11 @@ impl MyVerify {
actor_id
} else {
match self.0.fetch::<PublicKeyResponse>(&public_key_id).await {
match self
.0
.fetch::<PublicKeyResponse>(&public_key_id, BreakerStrategy::Require2XX)
.await
{
Ok(res) => res.actor_id().ok_or(ErrorKind::MissingId),
Err(e) => {
if e.is_gone() {
@ -85,7 +92,7 @@ impl MyVerify {
// Now we make sure we fetch an updated actor
let actor = self.1.get_no_cache(&actor_id, &self.0).await?;
do_verify(&actor.public_key, signature, signing_string).await?;
do_verify(&self.3, &actor.public_key, signature, signing_string).await?;
Ok(true)
}
@ -116,23 +123,29 @@ impl PublicKeyResponse {
#[tracing::instrument("Verify signature")]
async fn do_verify(
spawner: &Spawner,
public_key: &str,
signature: String,
signing_string: String,
) -> Result<(), Error> {
let public_key = RsaPublicKey::from_public_key_pem(public_key.trim())?;
let public_key_der = public_key
.to_pkcs1_der()
.map_err(|_| ErrorKind::DerEncode)?;
let public_key = ring::signature::UnparsedPublicKey::new(
&ring::signature::RSA_PKCS1_2048_8192_SHA256,
public_key_der,
);
let span = tracing::Span::current();
web::block(move || {
spawner
.spawn_blocking(move || {
span.in_scope(|| {
let decoded = STANDARD.decode(signature)?;
let signature =
Signature::try_from(decoded.as_slice()).map_err(ErrorKind::ReadSignature)?;
let verifying_key = VerifyingKey::<Sha256>::new(public_key);
verifying_key
.verify(signing_string.as_bytes(), &signature)
.map_err(ErrorKind::VerifySignature)?;
public_key
.verify(signing_string.as_bytes(), decoded.as_slice())
.map_err(|_| ErrorKind::VerifySignature)?;
Ok(()) as Result<(), Error>
})

2
src/middleware/webfinger.rs
View File

@ -1,10 +1,10 @@
use crate::{
config::{Config, UrlKind},
data::State,
future::LocalBoxFuture,
};
use actix_web::web::Data;
use actix_webfinger::{Resolver, Webfinger};
use futures_util::future::LocalBoxFuture;
use rsa_magic_public_key::AsMagicPublicKey;
pub(crate) struct RelayResolver;

246
src/requests.rs
View File

@ -1,31 +1,39 @@
use crate::{
data::LastOnline,
error::{Error, ErrorKind},
spawner::Spawner,
};
use activitystreams::iri_string::types::IriString;
use actix_web::http::header::Date;
use awc::{error::SendRequestError, Client, ClientResponse, Connector};
use base64::{engine::general_purpose::STANDARD, Engine};
use dashmap::DashMap;
use http_signature_normalization_actix::prelude::*;
use rand::thread_rng;
use rsa::{
pkcs1v15::SigningKey,
sha2::{Digest, Sha256},
signature::{RandomizedSigner, SignatureEncoding},
RsaPrivateKey,
use http_signature_normalization_reqwest::{digest::ring::Sha256, prelude::*};
use reqwest_middleware::ClientWithMiddleware;
use ring::{
rand::SystemRandom,
signature::{RsaKeyPair, RSA_PKCS1_SHA256},
};
use rsa::{pkcs1::EncodeRsaPrivateKey, RsaPrivateKey};
use std::{
sync::Arc,
time::{Duration, SystemTime},
};
use tracing_awc::Tracing;
const ONE_SECOND: u64 = 1;
const ONE_MINUTE: u64 = 60 * ONE_SECOND;
const ONE_HOUR: u64 = 60 * ONE_MINUTE;
const ONE_DAY: u64 = 24 * ONE_HOUR;
#[derive(Debug)]
pub(crate) enum BreakerStrategy {
// Requires a successful response
Require2XX,
// Allows HTTP 2xx-401
Allow401AndBelow,
// Allows HTTP 2xx-404
Allow404AndBelow,
}
#[derive(Clone)]
pub(crate) struct Breakers {
inner: Arc<DashMap<String, Breaker>>,
@ -38,7 +46,7 @@ impl std::fmt::Debug for Breakers {
}
impl Breakers {
fn should_try(&self, url: &IriString) -> bool {
pub(crate) fn should_try(&self, url: &IriString) -> bool {
if let Some(authority) = url.authority_str() {
if let Some(breaker) = self.inner.get(authority) {
breaker.should_try()
@ -140,12 +148,11 @@ impl Default for Breaker {
#[derive(Clone)]
pub(crate) struct Requests {
pool_size: usize,
client: Client,
client: ClientWithMiddleware,
key_id: String,
user_agent: String,
private_key: RsaPrivateKey,
config: Config,
private_key: Arc<RsaKeyPair>,
rng: SystemRandom,
config: Config<Spawner>,
breakers: Breakers,
last_online: Arc<LastOnline>,
}
@ -153,57 +160,42 @@ pub(crate) struct Requests {
impl std::fmt::Debug for Requests {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Requests")
.field("pool_size", &self.pool_size)
.field("key_id", &self.key_id)
.field("user_agent", &self.user_agent)
.field("config", &self.config)
.field("breakers", &self.breakers)
.finish()
}
}
thread_local! {
static CLIENT: std::cell::OnceCell<Client> = std::cell::OnceCell::new();
}
pub(crate) fn build_client(user_agent: &str, pool_size: usize) -> Client {
CLIENT.with(|client| {
client
.get_or_init(|| {
let connector = Connector::new().limit(pool_size);
Client::builder()
.connector(connector)
.wrap(Tracing)
.add_default_header(("User-Agent", user_agent.to_string()))
.timeout(Duration::from_secs(15))
.finish()
})
.clone()
})
}
impl Requests {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
key_id: String,
private_key: RsaPrivateKey,
user_agent: String,
breakers: Breakers,
last_online: Arc<LastOnline>,
pool_size: usize,
spawner: Spawner,
client: ClientWithMiddleware,
) -> Self {
let private_key_der = private_key.to_pkcs1_der().expect("Can encode der");
let private_key = ring::signature::RsaKeyPair::from_der(private_key_der.as_bytes())
.expect("Key is valid");
Requests {
pool_size,
client: build_client(&user_agent, pool_size),
client,
key_id,
user_agent,
private_key,
config: Config::default().mastodon_compat(),
private_key: Arc::new(private_key),
rng: SystemRandom::new(),
config: Config::new_with_spawner(spawner).mastodon_compat(),
breakers,
last_online,
}
}
pub(crate) fn spawner(mut self, spawner: Spawner) -> Self {
self.config = self.config.set_spawner(spawner);
self
}
pub(crate) fn reset_breaker(&self, iri: &IriString) {
self.breakers.succeed(iri);
}
@ -211,97 +203,126 @@ impl Requests {
async fn check_response(
&self,
parsed_url: &IriString,
res: Result<ClientResponse, SendRequestError>,
) -> Result<ClientResponse, Error> {
strategy: BreakerStrategy,
res: Result<reqwest::Response, reqwest_middleware::Error>,
) -> Result<reqwest::Response, Error> {
if res.is_err() {
self.breakers.fail(&parsed_url);
}
let mut res =
res.map_err(|e| ErrorKind::SendRequest(parsed_url.to_string(), e.to_string()))?;
let res = res?;
if res.status().is_server_error() {
let status = res.status();
let success = match strategy {
BreakerStrategy::Require2XX => status.is_success(),
BreakerStrategy::Allow401AndBelow => (200..=401).contains(&status.as_u16()),
BreakerStrategy::Allow404AndBelow => (200..=404).contains(&status.as_u16()),
};
if !success {
self.breakers.fail(&parsed_url);
if let Ok(bytes) = res.body().await {
if let Ok(s) = String::from_utf8(bytes.as_ref().to_vec()) {
if let Ok(s) = res.text().await {
if !s.is_empty() {
tracing::warn!("Response from {parsed_url}, {s}");
}
tracing::debug!("Response from {parsed_url}, {s}");
}
}
return Err(ErrorKind::Status(parsed_url.to_string(), res.status()).into());
return Err(ErrorKind::Status(parsed_url.to_string(), status).into());
}
// only actually succeed a breaker on 2xx response
if status.is_success() {
self.last_online.mark_seen(&parsed_url);
self.breakers.succeed(&parsed_url);
}
Ok(res)
}
#[tracing::instrument(name = "Fetch Json", skip(self), fields(signing_string))]
pub(crate) async fn fetch_json<T>(&self, url: &IriString) -> Result<T, Error>
pub(crate) async fn fetch_json<T>(
&self,
url: &IriString,
strategy: BreakerStrategy,
) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
self.do_fetch(url, "application/json").await
self.do_fetch(url, "application/json", strategy).await
}
#[tracing::instrument(name = "Fetch Json", skip(self), fields(signing_string))]
pub(crate) async fn fetch_json_msky<T>(&self, url: &IriString) -> Result<T, Error>
pub(crate) async fn fetch_json_msky<T>(
&self,
url: &IriString,
strategy: BreakerStrategy,
) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
let mut res = self
let body = self
.do_deliver(
url,
&serde_json::json!({}),
"application/json",
"application/json",
strategy,
)
.await?
.bytes()
.await?;
let body = res
.body()
.await
.map_err(|e| ErrorKind::ReceiveResponse(url.to_string(), e.to_string()))?;
Ok(serde_json::from_slice(body.as_ref())?)
Ok(serde_json::from_slice(&body)?)
}
#[tracing::instrument(name = "Fetch Activity+Json", skip(self), fields(signing_string))]
pub(crate) async fn fetch<T>(&self, url: &IriString) -> Result<T, Error>
pub(crate) async fn fetch<T>(
&self,
url: &IriString,
strategy: BreakerStrategy,
) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
self.do_fetch(url, "application/activity+json").await
self.do_fetch(url, "application/activity+json", strategy)
.await
}
async fn do_fetch<T>(&self, url: &IriString, accept: &str) -> Result<T, Error>
async fn do_fetch<T>(
&self,
url: &IriString,
accept: &str,
strategy: BreakerStrategy,
) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
let mut res = self.do_fetch_response(url, accept).await?;
let body = self
.do_fetch_response(url, accept, strategy)
.await?
.bytes()
.await?;
let body = res
.body()
.await
.map_err(|e| ErrorKind::ReceiveResponse(url.to_string(), e.to_string()))?;
Ok(serde_json::from_slice(body.as_ref())?)
Ok(serde_json::from_slice(&body)?)
}
#[tracing::instrument(name = "Fetch response", skip(self), fields(signing_string))]
pub(crate) async fn fetch_response(&self, url: &IriString) -> Result<ClientResponse, Error> {
self.do_fetch_response(url, "*/*").await
pub(crate) async fn fetch_response(
&self,
url: &IriString,
strategy: BreakerStrategy,
) -> Result<reqwest::Response, Error> {
self.do_fetch_response(url, "*/*", strategy).await
}
pub(crate) async fn do_fetch_response(
&self,
url: &IriString,
accept: &str,
) -> Result<ClientResponse, Error> {
strategy: BreakerStrategy,
) -> Result<reqwest::Response, Error> {
if !self.breakers.should_try(url) {
return Err(ErrorKind::Breaker.into());
}
@ -309,25 +330,20 @@ impl Requests {
let signer = self.signer();
let span = tracing::Span::current();
let res = self
let request = self
.client
.get(url.as_str())
.insert_header(("Accept", accept))
.insert_header(Date(SystemTime::now().into()))
.no_decompress()
.signature(
self.config.clone(),
self.key_id.clone(),
move |signing_string| {
.header("Accept", accept)
.header("Date", Date(SystemTime::now().into()).to_string())
.signature(&self.config, self.key_id.clone(), move |signing_string| {
span.record("signing_string", signing_string);
span.in_scope(|| signer.sign(signing_string))
},
)
.await?
.send()
.await;
})
.await?;
let res = self.check_response(url, res).await?;
let res = self.client.execute(request).await;
let res = self.check_response(url, strategy, res).await?;
Ok(res)
}
@ -337,7 +353,12 @@ impl Requests {
skip_all,
fields(inbox = inbox.to_string().as_str(), signing_string)
)]
pub(crate) async fn deliver<T>(&self, inbox: &IriString, item: &T) -> Result<(), Error>
pub(crate) async fn deliver<T>(
&self,
inbox: &IriString,
item: &T,
strategy: BreakerStrategy,
) -> Result<(), Error>
where
T: serde::ser::Serialize + std::fmt::Debug,
{
@ -346,6 +367,7 @@ impl Requests {
item,
"application/activity+json",
"application/activity+json",
strategy,
)
.await?;
Ok(())
@ -357,7 +379,8 @@ impl Requests {
item: &T,
content_type: &str,
accept: &str,
) -> Result<ClientResponse, Error>
strategy: BreakerStrategy,
) -> Result<reqwest::Response, Error>
where
T: serde::ser::Serialize + std::fmt::Debug,
{
@ -369,12 +392,12 @@ impl Requests {
let span = tracing::Span::current();
let item_string = serde_json::to_string(item)?;
let (req, body) = self
let request = self
.client
.post(inbox.as_str())
.insert_header(("Accept", accept))
.insert_header(("Content-Type", content_type))
.insert_header(Date(SystemTime::now().into()))
.header("Accept", accept)
.header("Content-Type", content_type)
.header("Date", Date(SystemTime::now().into()).to_string())
.signature_with_digest(
self.config.clone(),
self.key_id.clone(),
@ -385,12 +408,11 @@ impl Requests {
span.in_scope(|| signer.sign(signing_string))
},
)
.await?
.split();
.await?;
let res = req.send_body(body).await;
let res = self.client.execute(request).await;
let res = self.check_response(inbox, res).await?;
let res = self.check_response(inbox, strategy, res).await?;
Ok(res)
}
@ -398,19 +420,29 @@ impl Requests {
fn signer(&self) -> Signer {
Signer {
private_key: self.private_key.clone(),
rng: self.rng.clone(),
}
}
}
struct Signer {
private_key: RsaPrivateKey,
private_key: Arc<RsaKeyPair>,
rng: SystemRandom,
}
impl Signer {
fn sign(&self, signing_string: &str) -> Result<String, Error> {
let signing_key = SigningKey::<Sha256>::new(self.private_key.clone());
let signature =
signing_key.try_sign_with_rng(&mut thread_rng(), signing_string.as_bytes())?;
Ok(STANDARD.encode(signature.to_bytes().as_ref()))
let mut signature = vec![0; self.private_key.public_modulus_len()];
self.private_key
.sign(
&RSA_PKCS1_SHA256,
&self.rng,
signing_string.as_bytes(),
&mut signature,
)
.map_err(|_| ErrorKind::SignRequest)?;
Ok(STANDARD.encode(&signature))
}
}

6
src/routes/index.rs
View File

@ -36,12 +36,16 @@ pub(crate) async fn route(
state: web::Data<State>,
config: web::Data<Config>,
) -> Result<HttpResponse, Error> {
let all_nodes = state.node_cache().nodes().await?;
let all_nodes = state.node_cache.nodes().await?;
let mut nodes = Vec::new();
let mut local = Vec::new();
for node in all_nodes {
if !state.is_connected(&node.base) {
continue;
}
if node
.base
.authority_str()

12
src/routes/media.rs
View File

@ -1,4 +1,8 @@
use crate::{data::MediaCache, error::Error, requests::Requests};
use crate::{
data::MediaCache,
error::Error,
requests::{BreakerStrategy, Requests},
};
use actix_web::{body::BodyStream, web, HttpResponse};
use uuid::Uuid;
@ -11,7 +15,9 @@ pub(crate) async fn route(
let uuid = uuid.into_inner();
if let Some(url) = media.get_url(uuid).await? {
let res = requests.fetch_response(&url).await?;
let res = requests
.fetch_response(&url, BreakerStrategy::Allow404AndBelow)
.await?;
let mut response = HttpResponse::build(res.status());
@ -19,7 +25,7 @@ pub(crate) async fn route(
response.insert_header((name.clone(), value.clone()));
}
return Ok(response.body(BodyStream::new(res)));
return Ok(response.body(BodyStream::new(res.bytes_stream())));
}
Ok(HttpResponse::NotFound().finish())

193
src/spawner.rs Normal file
View File

@ -0,0 +1,193 @@
use http_signature_normalization_actix::{Canceled, Spawn};
use std::{
panic::AssertUnwindSafe,
sync::Arc,
thread::JoinHandle,
time::{Duration, Instant},
};
fn spawner_thread(
receiver: flume::Receiver<Box<dyn FnOnce() + Send>>,
name: &'static str,
id: usize,
) {
let guard = MetricsGuard::guard(name, id);
while let Ok(f) = receiver.recv() {
let start = Instant::now();
metrics::increment_counter!(format!("relay.{name}.operation.start"), "id" => id.to_string());
let res = std::panic::catch_unwind(AssertUnwindSafe(f));
metrics::increment_counter!(format!("relay.{name}.operation.end"), "complete" => res.is_ok().to_string(), "id" => id.to_string());
metrics::histogram!(format!("relay.{name}.operation.duration"), start.elapsed().as_secs_f64(), "complete" => res.is_ok().to_string(), "id" => id.to_string());
if let Err(e) = res {
tracing::warn!("{name} fn panicked: {e:?}");
}
}
guard.disarm();
}
#[derive(Clone, Debug)]
pub(crate) struct Spawner {
name: &'static str,
sender: Option<flume::Sender<Box<dyn FnOnce() + Send>>>,
threads: Option<Arc<Vec<JoinHandle<()>>>>,
}
struct MetricsGuard {
name: &'static str,
id: usize,
start: Instant,
armed: bool,
}
impl MetricsGuard {
fn guard(name: &'static str, id: usize) -> Self {
metrics::increment_counter!(format!("relay.{name}.launched"), "id" => id.to_string());
Self {
name,
id,
start: Instant::now(),
armed: true,
}
}
fn disarm(mut self) {
self.armed = false;
}
}
impl Drop for MetricsGuard {
fn drop(&mut self) {
metrics::increment_counter!(format!("relay.{}.closed", self.name), "clean" => (!self.armed).to_string(), "id" => self.id.to_string());
metrics::histogram!(format!("relay.{}.duration", self.name), self.start.elapsed().as_secs_f64(), "clean" => (!self.armed).to_string(), "id" => self.id.to_string());
tracing::warn!("Stopping {} - {}", self.name, self.id);
}
}
impl Spawner {
pub(crate) fn build(name: &'static str, threads: usize) -> std::io::Result<Self> {
let (sender, receiver) = flume::bounded(8);
tracing::warn!("Launching {threads} {name}s");
let threads = (0..threads)
.map(|i| {
let receiver = receiver.clone();
std::thread::Builder::new()
.name(format!("{name}-{i}"))
.spawn(move || {
spawner_thread(receiver, name, i);
})
})
.collect::<Result<Vec<_>, _>>()?;
Ok(Spawner {
name,
sender: Some(sender),
threads: Some(Arc::new(threads)),
})
}
}
impl Drop for Spawner {
fn drop(&mut self) {
self.sender.take();
if let Some(threads) = self.threads.take().and_then(Arc::into_inner) {
tracing::warn!("Joining {}s", self.name);
for thread in threads {
let _ = thread.join();
}
}
}
}
async fn timer<Fut>(fut: Fut) -> Fut::Output
where
Fut: std::future::Future,
{
let id = uuid::Uuid::new_v4();
metrics::increment_counter!("relay.spawner.wait-timer.start");
let mut interval = actix_rt::time::interval(Duration::from_secs(5));
// pass the first tick (instant)
interval.tick().await;
let mut fut = std::pin::pin!(fut);
let mut counter = 0;
loop {
tokio::select! {
out = &mut fut => {
metrics::increment_counter!("relay.spawner.wait-timer.end");
return out;
}
_ = interval.tick() => {
counter += 1;
metrics::increment_counter!("relay.spawner.wait-timer.pending");
tracing::warn!("Blocking operation {id} is taking a long time, {} seconds", counter * 5);
}
}
}
}
impl Spawn for Spawner {
type Future<T> = std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, Canceled>>>>;
fn spawn_blocking<Func, Out>(&self, func: Func) -> Self::Future<Out>
where
Func: FnOnce() -> Out + Send + 'static,
Out: Send + 'static,
{
let sender = self.sender.as_ref().expect("Sender exists").clone();
Box::pin(async move {
let (tx, rx) = flume::bounded(1);
let _ = sender
.send_async(Box::new(move || {
if tx.try_send((func)()).is_err() {
tracing::warn!("Requestor hung up");
metrics::increment_counter!("relay.spawner.disconnected");
}
}))
.await;
timer(rx.recv_async()).await.map_err(|_| Canceled)
})
}
}
impl http_signature_normalization_reqwest::Spawn for Spawner {
type Future<T> = std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, http_signature_normalization_reqwest::Canceled>> + Send>> where T: Send;
fn spawn_blocking<Func, Out>(&self, func: Func) -> Self::Future<Out>
where
Func: FnOnce() -> Out + Send + 'static,
Out: Send + 'static,
{
let sender = self.sender.as_ref().expect("Sender exists").clone();
Box::pin(async move {
let (tx, rx) = flume::bounded(1);
let _ = sender
.send_async(Box::new(move || {
if tx.try_send((func)()).is_err() {
tracing::warn!("Requestor hung up");
metrics::increment_counter!("relay.spawner.disconnected");
}
}))
.await;
timer(rx.recv_async())
.await
.map_err(|_| http_signature_normalization_reqwest::Canceled)
})
}
}