relay/src/requests.rs

429 lines
12 KiB
Rust
Raw Normal View History

2021-09-18 17:55:39 +00:00
use crate::error::{Error, ErrorKind};
2022-01-17 22:54:45 +00:00
use activitystreams::iri_string::types::IriString;
use actix_web::{http::header::Date, web::Bytes};
2021-03-10 02:09:56 +00:00
use awc::Client;
2021-11-23 22:19:59 +00:00
use dashmap::DashMap;
2020-03-18 04:35:20 +00:00
use http_signature_normalization_actix::prelude::*;
2022-10-29 17:22:13 +00:00
use rand::thread_rng;
use rsa::{pkcs1v15::SigningKey, RsaPrivateKey};
2020-03-18 04:35:20 +00:00
use sha2::{Digest, Sha256};
2022-10-29 17:22:13 +00:00
use signature::RandomizedSigner;
use std::{
cell::RefCell,
rc::Rc,
sync::{
atomic::{AtomicUsize, Ordering},
Arc,
},
2022-01-17 23:57:06 +00:00
time::{Duration, SystemTime},
};
2021-12-03 22:17:25 +00:00
use tracing_awc::Tracing;
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(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 {
2022-01-17 22:54:45 +00:00
fn should_try(&self, url: &IriString) -> bool {
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();
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 fn failure_threshold() -> usize {
10
}
2022-01-17 23:57:06 +00:00
fn failure_wait() -> Duration {
Duration::from_secs(ONE_DAY)
}
fn should_try(&self) -> bool {
self.failures < Self::failure_threshold()
2022-01-17 23:57:06 +00:00
|| 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: Rc<RefCell<Client>>,
consecutive_errors: Rc<AtomicUsize>,
error_limit: usize,
2020-03-18 04:35:20 +00:00
key_id: String,
user_agent: String,
2021-08-01 20:12:06 +00:00
private_key: RsaPrivateKey,
2020-03-18 04:35:20 +00:00
config: Config,
breakers: Breakers,
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("error_limit", &self.error_limit)
.field("key_id", &self.key_id)
.field("user_agent", &self.user_agent)
.field("config", &self.config)
.field("breakers", &self.breakers)
.finish()
}
}
2020-03-18 04:35:20 +00:00
impl Requests {
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,
user_agent: String,
breakers: Breakers,
) -> Self {
2020-03-18 04:35:20 +00:00
Requests {
client: Rc::new(RefCell::new(
2020-09-14 00:46:13 +00:00
Client::builder()
2021-12-03 22:17:25 +00:00
.wrap(Tracing)
.add_default_header(("User-Agent", user_agent.clone()))
.finish(),
)),
consecutive_errors: Rc::new(AtomicUsize::new(0)),
error_limit: 3,
2020-03-18 04:35:20 +00:00
key_id,
user_agent,
2020-03-18 04:35:20 +00:00
private_key,
2020-09-30 00:33:50 +00:00
config: Config::default().mastodon_compat(),
breakers,
2020-03-18 04:35:20 +00:00
}
}
fn count_err(&self) {
let count = self.consecutive_errors.fetch_add(1, Ordering::Relaxed);
if count + 1 >= self.error_limit {
2022-11-16 01:56:13 +00:00
tracing::warn!("{} consecutive errors, rebuilding http client", count);
2020-09-14 00:46:13 +00:00
*self.client.borrow_mut() = Client::builder()
2021-12-03 22:17:25 +00:00
.wrap(Tracing)
.add_default_header(("User-Agent", self.user_agent.clone()))
.finish();
self.reset_err();
}
}
fn reset_err(&self) {
self.consecutive_errors.swap(0, Ordering::Relaxed);
}
2022-11-01 20:57:33 +00:00
#[tracing::instrument(name = "Fetch Json", skip(self))]
2021-09-18 17:55:39 +00:00
pub(crate) async fn fetch_json<T>(&self, url: &str) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
self.do_fetch(url, "application/json").await
}
2022-11-01 20:57:33 +00:00
#[tracing::instrument(name = "Fetch Activity+Json", skip(self))]
2021-09-18 17:55:39 +00:00
pub(crate) async fn fetch<T>(&self, url: &str) -> Result<T, Error>
where
T: serde::de::DeserializeOwned,
{
self.do_fetch(url, "application/activity+json").await
}
2021-09-18 17:55:39 +00:00
async fn do_fetch<T>(&self, url: &str, accept: &str) -> Result<T, Error>
2020-03-18 04:35:20 +00:00
where
T: serde::de::DeserializeOwned,
{
2022-01-17 22:54:45 +00:00
let parsed_url = url.parse::<IriString>()?;
2021-11-23 22:19:59 +00:00
if !self.breakers.should_try(&parsed_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 client: Client = self.client.borrow().clone();
2020-09-07 21:51:02 +00:00
let res = client
.get(url)
2021-02-11 00:00:11 +00:00
.insert_header(("Accept", accept))
.insert_header(Date(SystemTime::now().into()))
2020-03-30 06:06:13 +00:00
.signature(
self.config.clone(),
self.key_id.clone(),
move |signing_string| signer.sign(signing_string),
)
.await?
2020-03-18 04:35:20 +00:00
.send()
.await;
if res.is_err() {
self.count_err();
2021-11-23 22:19:59 +00:00
self.breakers.fail(&parsed_url);
}
2021-09-18 17:55:39 +00:00
let mut res = res.map_err(|e| ErrorKind::SendRequest(url.to_string(), e.to_string()))?;
self.reset_err();
2020-03-18 04:35:20 +00:00
if !res.status().is_success() {
if let Ok(bytes) = res.body().await {
if let Ok(s) = String::from_utf8(bytes.as_ref().to_vec()) {
if !s.is_empty() {
2022-11-16 01:56:13 +00:00
tracing::warn!("Response from {}, {}", url, s);
}
2020-03-18 04:35:20 +00:00
}
}
2020-03-18 04:35:20 +00:00
2021-11-23 22:19:59 +00:00
self.breakers.fail(&parsed_url);
2021-09-18 17:55:39 +00:00
return Err(ErrorKind::Status(url.to_string(), res.status()).into());
}
2021-11-23 22:19:59 +00:00
self.breakers.succeed(&parsed_url);
let body = res
.body()
.await
2021-09-18 17:55:39 +00:00
.map_err(|e| ErrorKind::ReceiveResponse(url.to_string(), e.to_string()))?;
Ok(serde_json::from_slice(body.as_ref())?)
}
2022-11-01 20:57:33 +00:00
#[tracing::instrument(name = "Fetch Bytes", skip(self))]
2021-09-18 17:55:39 +00:00
pub(crate) async fn fetch_bytes(&self, url: &str) -> Result<(String, Bytes), Error> {
2022-01-17 22:54:45 +00:00
let parsed_url = url.parse::<IriString>()?;
2021-11-23 22:19:59 +00:00
if !self.breakers.should_try(&parsed_url) {
2021-09-18 17:55:39 +00:00
return Err(ErrorKind::Breaker.into());
}
2022-11-16 01:56:13 +00:00
tracing::info!("Fetching bytes for {}", url);
2020-03-30 06:06:13 +00:00
let signer = self.signer();
let client: Client = self.client.borrow().clone();
2020-09-07 21:51:02 +00:00
let res = client
.get(url)
2021-02-11 00:00:11 +00:00
.insert_header(("Accept", "*/*"))
.insert_header(Date(SystemTime::now().into()))
2020-03-30 06:06:13 +00:00
.signature(
self.config.clone(),
self.key_id.clone(),
move |signing_string| signer.sign(signing_string),
)
.await?
.send()
.await;
if res.is_err() {
2021-11-23 22:19:59 +00:00
self.breakers.fail(&parsed_url);
self.count_err();
}
2021-09-18 17:55:39 +00:00
let mut res = res.map_err(|e| ErrorKind::SendRequest(url.to_string(), e.to_string()))?;
self.reset_err();
let content_type = if let Some(content_type) = res.headers().get("content-type") {
if let Ok(s) = content_type.to_str() {
s.to_owned()
} else {
2021-09-18 17:55:39 +00:00
return Err(ErrorKind::ContentType.into());
}
} else {
2021-09-18 17:55:39 +00:00
return Err(ErrorKind::ContentType.into());
};
if !res.status().is_success() {
if let Ok(bytes) = res.body().await {
if let Ok(s) = String::from_utf8(bytes.as_ref().to_vec()) {
if !s.is_empty() {
2022-11-16 01:56:13 +00:00
tracing::warn!("Response from {}, {}", url, s);
}
}
}
2021-11-23 22:19:59 +00:00
self.breakers.fail(&parsed_url);
2021-09-18 17:55:39 +00:00
return Err(ErrorKind::Status(url.to_string(), res.status()).into());
}
2021-11-23 22:19:59 +00:00
self.breakers.succeed(&parsed_url);
let bytes = match res.body().limit(1024 * 1024 * 4).await {
Err(e) => {
2021-09-18 17:55:39 +00:00
return Err(ErrorKind::ReceiveResponse(url.to_string(), e.to_string()).into());
}
Ok(bytes) => bytes,
};
Ok((content_type, bytes))
}
2021-09-21 19:32:25 +00:00
#[tracing::instrument(
"Deliver to Inbox",
2022-11-01 20:57:33 +00:00
skip_all,
2022-11-13 19:57:52 +00:00
fields(inbox = inbox.to_string().as_str())
2021-09-21 19:32:25 +00:00
)]
2022-01-17 22:54:45 +00:00
pub(crate) async fn deliver<T>(&self, inbox: IriString, item: &T) -> Result<(), 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();
2020-03-18 04:35:20 +00:00
let item_string = serde_json::to_string(item)?;
let client: Client = self.client.borrow().clone();
2021-09-21 16:21:06 +00:00
let (req, body) = client
2020-09-07 21:51:02 +00:00
.post(inbox.as_str())
2021-02-11 00:00:11 +00:00
.insert_header(("Accept", "application/activity+json"))
.insert_header(("Content-Type", "application/activity+json"))
.insert_header(Date(SystemTime::now().into()))
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,
2020-03-30 06:06:13 +00:00
move |signing_string| signer.sign(signing_string),
)
.await?
2021-09-21 16:21:06 +00:00
.split();
2021-12-03 22:17:25 +00:00
let res = req.send_body(body).await;
if res.is_err() {
self.count_err();
2021-11-23 22:19:59 +00:00
self.breakers.fail(&inbox);
}
2021-09-18 17:55:39 +00:00
let mut res = res.map_err(|e| ErrorKind::SendRequest(inbox.to_string(), e.to_string()))?;
self.reset_err();
2020-03-18 04:35:20 +00:00
if !res.status().is_success() {
if let Ok(bytes) = res.body().await {
if let Ok(s) = String::from_utf8(bytes.as_ref().to_vec()) {
if !s.is_empty() {
tracing::warn!("Response from {}, {}", inbox.as_str(), s);
}
2020-03-18 04:35:20 +00:00
}
2022-11-08 19:50:08 +00:00
}
self.breakers.fail(&inbox);
return Err(ErrorKind::Status(inbox.to_string(), res.status()).into());
2020-03-18 04:35:20 +00:00
}
2021-11-23 22:19:59 +00:00
self.breakers.succeed(&inbox);
2020-03-18 04:35:20 +00:00
Ok(())
}
2020-03-30 06:06:13 +00:00
fn signer(&self) -> Signer {
Signer {
private_key: self.private_key.clone(),
}
}
}
struct Signer {
2021-08-01 20:12:06 +00:00
private_key: RsaPrivateKey,
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> {
2022-10-29 17:22:13 +00:00
let signing_key = SigningKey::<Sha256>::new_with_prefix(self.private_key.clone());
2022-10-29 18:47:23 +00:00
let signature = signing_key.try_sign_with_rng(thread_rng(), signing_string.as_bytes())?;
2022-10-29 17:22:13 +00:00
Ok(base64::encode(signature.as_ref()))
2020-03-18 04:35:20 +00:00
}
}