relay/src/requests.rs

449 lines
12 KiB
Rust
Raw Normal View History

use crate::{
data::LastOnline,
error::{Error, ErrorKind},
2023-07-27 15:19:20 +00:00
spawner::Spawner,
};
2022-01-17 22:54:45 +00:00
use activitystreams::iri_string::types::IriString;
2022-11-18 04:39:26 +00:00
use actix_web::http::header::Date;
2023-01-23 14:38:55 +00:00
use base64::{engine::general_purpose::STANDARD, Engine};
2021-11-23 22:19:59 +00:00
use dashmap::DashMap;
use http_signature_normalization_reqwest::{digest::ring::Sha256, prelude::*};
use reqwest_middleware::ClientWithMiddleware;
2023-08-04 23:34:42 +00:00
use ring::{
rand::SystemRandom,
signature::{RsaKeyPair, RSA_PKCS1_SHA256},
2023-01-23 14:56:18 +00:00
};
2023-08-04 23:34:42 +00:00
use rsa::{pkcs1::EncodeRsaPrivateKey, RsaPrivateKey};
use std::{
2023-07-27 15:19:20 +00:00
sync::Arc,
time::{Duration, SystemTime},
};
2020-03-18 04:35:20 +00:00
2022-01-17 23:57:06 +00:00
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)]
2021-02-10 04:17:20 +00:00
pub(crate) struct Breakers {
2021-11-23 22:19:59 +00:00
inner: Arc<DashMap<String, Breaker>>,
}
2021-09-18 17:55:39 +00:00
impl std::fmt::Debug for Breakers {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Breakers").finish()
}
}
impl Breakers {
pub(crate) fn should_try(&self, url: &IriString) -> bool {
2022-01-17 22:54:45 +00:00
if let Some(authority) = url.authority_str() {
if let Some(breaker) = self.inner.get(authority) {
2021-11-23 22:19:59 +00:00
breaker.should_try()
} else {
true
}
} else {
false
}
}
2022-01-17 22:54:45 +00:00
fn fail(&self, url: &IriString) {
if let Some(authority) = url.authority_str() {
let should_write = {
2022-01-17 22:54:45 +00:00
if let Some(mut breaker) = self.inner.get_mut(authority) {
2021-11-23 22:19:59 +00:00
breaker.fail();
if !breaker.should_try() {
tracing::warn!("Failed breaker for {authority}");
}
false
} else {
true
}
};
if should_write {
2022-01-17 22:54:45 +00:00
let mut breaker = self.inner.entry(authority.to_owned()).or_default();
2021-11-23 22:19:59 +00:00
breaker.fail();
}
}
}
2022-01-17 22:54:45 +00:00
fn succeed(&self, url: &IriString) {
if let Some(authority) = url.authority_str() {
let should_write = {
2022-01-17 22:54:45 +00:00
if let Some(mut breaker) = self.inner.get_mut(authority) {
2021-11-23 22:19:59 +00:00
breaker.succeed();
false
} else {
true
}
};
if should_write {
2022-01-17 22:54:45 +00:00
let mut breaker = self.inner.entry(authority.to_owned()).or_default();
2021-11-23 22:19:59 +00:00
breaker.succeed();
}
}
}
}
impl Default for Breakers {
fn default() -> Self {
Breakers {
2021-11-23 22:19:59 +00:00
inner: Arc::new(DashMap::new()),
}
}
}
2021-09-18 17:55:39 +00:00
#[derive(Debug)]
struct Breaker {
failures: usize,
2022-01-17 23:57:06 +00:00
last_attempt: SystemTime,
last_success: SystemTime,
}
impl Breaker {
const FAILURE_WAIT: Duration = Duration::from_secs(ONE_DAY);
const FAILURE_THRESHOLD: usize = 10;
fn should_try(&self) -> bool {
self.failures < Self::FAILURE_THRESHOLD
|| self.last_attempt + Self::FAILURE_WAIT < SystemTime::now()
}
fn fail(&mut self) {
self.failures += 1;
2022-01-17 23:57:06 +00:00
self.last_attempt = SystemTime::now();
}
fn succeed(&mut self) {
self.failures = 0;
2022-01-17 23:57:06 +00:00
self.last_attempt = SystemTime::now();
self.last_success = SystemTime::now();
}
}
impl Default for Breaker {
fn default() -> Self {
2022-01-17 23:57:06 +00:00
let now = SystemTime::now();
Breaker {
failures: 0,
last_attempt: now,
last_success: now,
}
}
}
2020-03-18 04:35:20 +00:00
#[derive(Clone)]
2021-02-10 04:17:20 +00:00
pub(crate) struct Requests {
client: ClientWithMiddleware,
2020-03-18 04:35:20 +00:00
key_id: String,
2023-08-04 23:34:42 +00:00
private_key: Arc<RsaKeyPair>,
rng: SystemRandom,
config: Config<Spawner>,
breakers: Breakers,
last_online: Arc<LastOnline>,
2020-03-18 04:35:20 +00:00
}
2021-09-18 17:55:39 +00:00
impl std::fmt::Debug for Requests {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Requests")
.field("key_id", &self.key_id)
.field("config", &self.config)
.field("breakers", &self.breakers)
.finish()
}
}
2020-03-18 04:35:20 +00:00
impl Requests {
2023-07-26 23:04:09 +00:00
#[allow(clippy::too_many_arguments)]
2021-02-10 04:17:20 +00:00
pub(crate) fn new(
key_id: String,
2021-08-01 20:12:06 +00:00
private_key: RsaPrivateKey,
breakers: Breakers,
last_online: Arc<LastOnline>,
spawner: Spawner,
client: ClientWithMiddleware,
) -> Self {
2023-08-04 23:34:42 +00:00
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");
2020-03-18 04:35:20 +00:00
Requests {
client,
2020-03-18 04:35:20 +00:00
key_id,
2023-08-04 23:34:42 +00:00
private_key: Arc::new(private_key),
rng: SystemRandom::new(),
config: Config::new_with_spawner(spawner).mastodon_compat(),
breakers,
last_online,
2020-03-18 04:35:20 +00:00
}
}
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);
}
async fn check_response(
&self,
parsed_url: &IriString,
strategy: BreakerStrategy,
res: Result<reqwest::Response, reqwest_middleware::Error>,
) -> Result<reqwest::Response, Error> {
if res.is_err() {
self.breakers.fail(&parsed_url);
}
let res = res?;
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(s) = res.text().await {
if !s.is_empty() {
tracing::debug!("Response from {parsed_url}, {s}");
}
}
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,
strategy: BreakerStrategy,
) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
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,
strategy: BreakerStrategy,
) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
let body = self
.do_deliver(
url,
&serde_json::json!({}),
"application/json",
"application/json",
strategy,
)
.await?
.bytes()
.await?;
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,
strategy: BreakerStrategy,
) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
self.do_fetch(url, "application/activity+json", strategy)
.await
}
async fn do_fetch<T>(
&self,
url: &IriString,
accept: &str,
strategy: BreakerStrategy,
) -> Result<T, Error>
2020-03-18 04:35:20 +00:00
where
T: serde::de::DeserializeOwned,
{
let body = self
.do_fetch_response(url, accept, strategy)
.await?
.bytes()
.await?;
Ok(serde_json::from_slice(&body)?)
}
2022-11-18 04:39:26 +00:00
#[tracing::instrument(name = "Fetch response", skip(self), fields(signing_string))]
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,
strategy: BreakerStrategy,
) -> Result<reqwest::Response, Error> {
if !self.breakers.should_try(url) {
2021-09-18 17:55:39 +00:00
return Err(ErrorKind::Breaker.into());
}
2020-03-30 06:06:13 +00:00
let signer = self.signer();
let span = tracing::Span::current();
2020-03-30 06:06:13 +00:00
let request = self
.client
2022-11-18 04:39:26 +00:00
.get(url.as_str())
.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?;
let res = self.client.execute(request).await;
let res = self.check_response(url, strategy, res).await?;
2022-11-18 04:39:26 +00:00
Ok(res)
}
2021-09-21 19:32:25 +00:00
#[tracing::instrument(
"Deliver to Inbox",
2022-11-01 20:57:33 +00:00
skip_all,
fields(inbox = inbox.to_string().as_str(), signing_string)
2021-09-21 19:32:25 +00:00
)]
pub(crate) async fn deliver<T>(
&self,
inbox: &IriString,
item: &T,
strategy: BreakerStrategy,
) -> Result<(), Error>
where
T: serde::ser::Serialize + std::fmt::Debug,
{
self.do_deliver(
inbox,
item,
"application/activity+json",
"application/activity+json",
strategy,
)
.await?;
Ok(())
}
async fn do_deliver<T>(
&self,
inbox: &IriString,
item: &T,
content_type: &str,
accept: &str,
strategy: BreakerStrategy,
) -> Result<reqwest::Response, Error>
2020-03-18 04:35:20 +00:00
where
2021-09-21 18:26:31 +00:00
T: serde::ser::Serialize + std::fmt::Debug,
2020-03-18 04:35:20 +00:00
{
2021-11-23 22:19:59 +00:00
if !self.breakers.should_try(&inbox) {
2021-09-18 17:55:39 +00:00
return Err(ErrorKind::Breaker.into());
}
2020-03-30 06:06:13 +00:00
let signer = self.signer();
let span = tracing::Span::current();
2020-03-18 04:35:20 +00:00
let item_string = serde_json::to_string(item)?;
let request = self
.client
2020-09-07 21:51:02 +00:00
.post(inbox.as_str())
.header("Accept", accept)
.header("Content-Type", content_type)
.header("Date", Date(SystemTime::now().into()).to_string())
2020-03-18 04:35:20 +00:00
.signature_with_digest(
2020-03-30 06:06:13 +00:00
self.config.clone(),
self.key_id.clone(),
Sha256::new(),
2020-03-18 04:35:20 +00:00
item_string,
move |signing_string| {
span.record("signing_string", signing_string);
span.in_scope(|| signer.sign(signing_string))
},
2020-03-30 06:06:13 +00:00
)
.await?;
2021-09-21 16:21:06 +00:00
let res = self.client.execute(request).await;
let res = self.check_response(inbox, strategy, res).await?;
Ok(res)
2020-03-18 04:35:20 +00:00
}
2020-03-30 06:06:13 +00:00
fn signer(&self) -> Signer {
Signer {
private_key: self.private_key.clone(),
2023-08-04 23:34:42 +00:00
rng: self.rng.clone(),
2020-03-30 06:06:13 +00:00
}
}
}
struct Signer {
2023-08-04 23:34:42 +00:00
private_key: Arc<RsaKeyPair>,
rng: SystemRandom,
2020-03-30 06:06:13 +00:00
}
impl Signer {
2021-09-18 17:55:39 +00:00
fn sign(&self, signing_string: &str) -> Result<String, Error> {
let mut signature = vec![0; self.private_key.public().modulus_len()];
2023-08-04 23:34:42 +00:00
self.private_key
.sign(
&RSA_PKCS1_SHA256,
&self.rng,
signing_string.as_bytes(),
&mut signature,
)
.map_err(|_| ErrorKind::SignRequest)?;
Ok(STANDARD.encode(&signature))
2020-03-18 04:35:20 +00:00
}
}