mirror of
https://git.asonix.dog/asonix/relay.git
synced 2024-06-26 00:50:34 +00:00
Compare commits
11 commits
Author | SHA1 | Date | |
---|---|---|---|
cda92e7523 | |||
43b03a176c | |||
a465d1ae5b | |||
4fa7674a35 | |||
8c14d613f7 | |||
aff2431681 | |||
5aa97212b3 | |||
97567cf598 | |||
4c663f399e | |||
8a3256f52a | |||
13a2653fe8 |
|
@ -21,7 +21,8 @@ jobs:
|
|||
-
|
||||
name: Clippy
|
||||
run: |
|
||||
cargo clippy --no-default-features -- -D warnings
|
||||
# cargo clippy --no-default-features -- -D warnings
|
||||
cargo clippy --no-default-features
|
||||
|
||||
tests:
|
||||
runs-on: docker
|
||||
|
@ -207,7 +208,6 @@ jobs:
|
|||
direction: upload
|
||||
token: ${{ secrets.GITHUB_TOKEN }}
|
||||
release-dir: artifacts/
|
||||
prerelease: true
|
||||
|
||||
publish-crate:
|
||||
needs: [build]
|
||||
|
|
510
Cargo.lock
generated
510
Cargo.lock
generated
File diff suppressed because it is too large
Load diff
|
@ -1,7 +1,7 @@
|
|||
[package]
|
||||
name = "ap-relay"
|
||||
description = "A simple activitypub relay"
|
||||
version = "0.3.112"
|
||||
version = "0.3.114"
|
||||
authors = ["asonix <asonix@asonix.dog>"]
|
||||
license = "AGPL-3.0"
|
||||
readme = "README.md"
|
||||
|
@ -38,6 +38,7 @@ config = { version = "0.14.0", default-features = false, features = ["toml", "js
|
|||
console-subscriber = { version = "0.2", optional = true }
|
||||
dashmap = "5.1.0"
|
||||
dotenv = "0.15.0"
|
||||
futures-core = "0.3.30"
|
||||
lru = "0.12.0"
|
||||
metrics = "0.22.0"
|
||||
metrics-exporter-prometheus = { version = "0.13.0", default-features = false, features = [
|
||||
|
|
|
@ -20,11 +20,11 @@
|
|||
},
|
||||
"nixpkgs": {
|
||||
"locked": {
|
||||
"lastModified": 1711163522,
|
||||
"narHash": "sha256-YN/Ciidm+A0fmJPWlHBGvVkcarYWSC+s3NTPk/P+q3c=",
|
||||
"lastModified": 1719075281,
|
||||
"narHash": "sha256-CyyxvOwFf12I91PBWz43iGT1kjsf5oi6ax7CrvaMyAo=",
|
||||
"owner": "NixOS",
|
||||
"repo": "nixpkgs",
|
||||
"rev": "44d0940ea560dee511026a53f0e2e2cde489b4d4",
|
||||
"rev": "a71e967ef3694799d0c418c98332f7ff4cc5f6af",
|
||||
"type": "github"
|
||||
},
|
||||
"original": {
|
||||
|
|
|
@ -5,7 +5,7 @@
|
|||
|
||||
rustPlatform.buildRustPackage {
|
||||
pname = "relay";
|
||||
version = "0.3.112";
|
||||
version = "0.3.114";
|
||||
src = ./.;
|
||||
cargoLock.lockFile = ./Cargo.lock;
|
||||
|
||||
|
|
|
@ -605,6 +605,8 @@ impl Db {
|
|||
|
||||
pub(crate) async fn add_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
|
||||
self.unblock(move |inner| {
|
||||
let connected_by_domain = inner.connected_by_domain(&domains).collect::<Vec<_>>();
|
||||
|
||||
let res = (
|
||||
&inner.connected_actor_ids,
|
||||
&inner.blocked_domains,
|
||||
|
@ -615,7 +617,7 @@ impl Db {
|
|||
let mut blocked_batch = Batch::default();
|
||||
let mut allowed_batch = Batch::default();
|
||||
|
||||
for connected in inner.connected_by_domain(&domains) {
|
||||
for connected in &connected_by_domain {
|
||||
connected_batch.remove(connected.as_str().as_bytes());
|
||||
}
|
||||
|
||||
|
@ -687,9 +689,11 @@ impl Db {
|
|||
pub(crate) async fn remove_allows(&self, domains: Vec<String>) -> Result<(), Error> {
|
||||
self.unblock(move |inner| {
|
||||
if inner.restricted_mode {
|
||||
let connected_by_domain = inner.connected_by_domain(&domains).collect::<Vec<_>>();
|
||||
|
||||
let mut connected_batch = Batch::default();
|
||||
|
||||
for connected in inner.connected_by_domain(&domains) {
|
||||
for connected in &connected_by_domain {
|
||||
connected_batch.remove(connected.as_str().as_bytes());
|
||||
}
|
||||
|
||||
|
|
|
@ -123,6 +123,9 @@ pub(crate) enum ErrorKind {
|
|||
#[error("Couldn't sign request")]
|
||||
SignRequest,
|
||||
|
||||
#[error("Response body from server exceeded limits")]
|
||||
BodyTooLarge,
|
||||
|
||||
#[error("Couldn't make request")]
|
||||
Reqwest(#[from] reqwest::Error),
|
||||
|
||||
|
|
|
@ -156,7 +156,7 @@ struct Link {
|
|||
#[serde(untagged)]
|
||||
enum MaybeSupported<T> {
|
||||
Supported(T),
|
||||
Unsupported(String),
|
||||
Unsupported(#[allow(unused)] String),
|
||||
}
|
||||
|
||||
impl<T> MaybeSupported<T> {
|
||||
|
@ -165,8 +165,8 @@ impl<T> MaybeSupported<T> {
|
|||
}
|
||||
}
|
||||
|
||||
struct SupportedVersion(String);
|
||||
struct SupportedNodeinfo(String);
|
||||
struct SupportedVersion(#[allow(unused)] String);
|
||||
struct SupportedNodeinfo(#[allow(unused)] String);
|
||||
|
||||
static SUPPORTED_VERSIONS: &str = "2.";
|
||||
static SUPPORTED_NODEINFO: &str = "http://nodeinfo.diaspora.software/ns/schema/2.";
|
||||
|
|
|
@ -38,6 +38,7 @@ mod middleware;
|
|||
mod requests;
|
||||
mod routes;
|
||||
mod spawner;
|
||||
mod stream;
|
||||
mod telegram;
|
||||
|
||||
use crate::config::UrlKind;
|
||||
|
|
|
@ -2,6 +2,7 @@ use crate::{
|
|||
data::LastOnline,
|
||||
error::{Error, ErrorKind},
|
||||
spawner::Spawner,
|
||||
stream::{aggregate, limit_stream},
|
||||
};
|
||||
use activitystreams::iri_string::types::IriString;
|
||||
use actix_web::http::header::Date;
|
||||
|
@ -24,6 +25,9 @@ const ONE_MINUTE: u64 = 60 * ONE_SECOND;
|
|||
const ONE_HOUR: u64 = 60 * ONE_MINUTE;
|
||||
const ONE_DAY: u64 = 24 * ONE_HOUR;
|
||||
|
||||
// 20 KB
|
||||
const JSON_SIZE_LIMIT: usize = 20 * 1024;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) enum BreakerStrategy {
|
||||
// Requires a successful response
|
||||
|
@ -262,7 +266,7 @@ impl Requests {
|
|||
where
|
||||
T: serde::de::DeserializeOwned,
|
||||
{
|
||||
let body = self
|
||||
let stream = self
|
||||
.do_deliver(
|
||||
url,
|
||||
&serde_json::json!({}),
|
||||
|
@ -271,8 +275,9 @@ impl Requests {
|
|||
strategy,
|
||||
)
|
||||
.await?
|
||||
.bytes()
|
||||
.await?;
|
||||
.bytes_stream();
|
||||
|
||||
let body = aggregate(limit_stream(stream, JSON_SIZE_LIMIT)).await?;
|
||||
|
||||
Ok(serde_json::from_slice(&body)?)
|
||||
}
|
||||
|
@ -299,11 +304,12 @@ impl Requests {
|
|||
where
|
||||
T: serde::de::DeserializeOwned,
|
||||
{
|
||||
let body = self
|
||||
let stream = self
|
||||
.do_fetch_response(url, accept, strategy)
|
||||
.await?
|
||||
.bytes()
|
||||
.await?;
|
||||
.bytes_stream();
|
||||
|
||||
let body = aggregate(limit_stream(stream, JSON_SIZE_LIMIT)).await?;
|
||||
|
||||
Ok(serde_json::from_slice(&body)?)
|
||||
}
|
||||
|
|
|
@ -2,10 +2,14 @@ use crate::{
|
|||
data::MediaCache,
|
||||
error::Error,
|
||||
requests::{BreakerStrategy, Requests},
|
||||
stream::limit_stream,
|
||||
};
|
||||
use actix_web::{body::BodyStream, web, HttpResponse};
|
||||
use uuid::Uuid;
|
||||
|
||||
// 16 MB
|
||||
const IMAGE_SIZE_LIMIT: usize = 16 * 1024 * 1024;
|
||||
|
||||
#[tracing::instrument(name = "Media", skip(media, requests))]
|
||||
pub(crate) async fn route(
|
||||
media: web::Data<MediaCache>,
|
||||
|
@ -25,7 +29,10 @@ pub(crate) async fn route(
|
|||
response.insert_header((name.clone(), value.clone()));
|
||||
}
|
||||
|
||||
return Ok(response.body(BodyStream::new(res.bytes_stream())));
|
||||
return Ok(response.body(BodyStream::new(limit_stream(
|
||||
res.bytes_stream(),
|
||||
IMAGE_SIZE_LIMIT,
|
||||
))));
|
||||
}
|
||||
|
||||
Ok(HttpResponse::NotFound().finish())
|
||||
|
|
59
src/stream.rs
Normal file
59
src/stream.rs
Normal file
|
@ -0,0 +1,59 @@
|
|||
use crate::error::{Error, ErrorKind};
|
||||
use actix_web::web::{Bytes, BytesMut};
|
||||
use futures_core::Stream;
|
||||
use streem::IntoStreamer;
|
||||
|
||||
pub(crate) fn limit_stream<'a, S>(
|
||||
input: S,
|
||||
limit: usize,
|
||||
) -> impl Stream<Item = Result<Bytes, Error>> + Send + 'a
|
||||
where
|
||||
S: Stream<Item = reqwest::Result<Bytes>> + Send + 'a,
|
||||
{
|
||||
streem::try_from_fn(move |yielder| async move {
|
||||
let stream = std::pin::pin!(input);
|
||||
let mut stream = stream.into_streamer();
|
||||
|
||||
let mut count = 0;
|
||||
|
||||
while let Some(bytes) = stream.try_next().await? {
|
||||
count += bytes.len();
|
||||
|
||||
if count > limit {
|
||||
return Err(ErrorKind::BodyTooLarge.into());
|
||||
}
|
||||
|
||||
yielder.yield_ok(bytes).await;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) async fn aggregate<S>(input: S) -> Result<Bytes, Error>
|
||||
where
|
||||
S: Stream<Item = Result<Bytes, Error>>,
|
||||
{
|
||||
let stream = std::pin::pin!(input);
|
||||
let mut streamer = stream.into_streamer();
|
||||
|
||||
let mut buf = Vec::new();
|
||||
|
||||
while let Some(bytes) = streamer.try_next().await? {
|
||||
buf.push(bytes);
|
||||
}
|
||||
|
||||
if buf.len() == 1 {
|
||||
return Ok(buf.pop().expect("buf has exactly one element"));
|
||||
}
|
||||
|
||||
let total_size: usize = buf.iter().map(|b| b.len()).sum();
|
||||
|
||||
let mut bytes_mut = BytesMut::with_capacity(total_size);
|
||||
|
||||
for bytes in &buf {
|
||||
bytes_mut.extend_from_slice(&bytes);
|
||||
}
|
||||
|
||||
Ok(bytes_mut.freeze())
|
||||
}
|
Loading…
Reference in a new issue