Prepare 0.3.116

Reviewed-on: https://git.asonix.dog/asonix/relay/pulls/3

.forgejo/workflows/publish.yaml

@@ -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]

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "ap-relay"
 description = "A simple activitypub relay"
-version = "0.3.112"
+version = "0.3.116"
 authors = ["asonix <asonix@asonix.dog>"]
 license = "AGPL-3.0"
 readme = "README.md"
@@ -24,59 +24,60 @@ default = []
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-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_23"] }
 actix-webfinger = { version = "0.5.0", default-features = false }
 activitystreams = "0.7.0-alpha.25"
 activitystreams-ext = "0.1.0-alpha.3"
 ammonia = "4.0.0"
-async-cpupool = "0.2.2"
-bcrypt = "0.15"
+async-cpupool = "0.3.0"
+bcrypt = "0.16"
 base64 = "0.22"
 clap = { version = "4.0.0", features = ["derive"] }
 color-eyre = "0.6.2"
 config = { version = "0.14.0", default-features = false, features = ["toml", "json", "yaml"] }
-console-subscriber = { version = "0.2", optional = true }
-dashmap = "5.1.0"
+console-subscriber = { version = "0.4", optional = true }
+dashmap = "6.0.1"
 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 = [
+metrics = "0.23.0"
+metrics-exporter-prometheus = { version = "0.15.0", default-features = false, features = [
   "http-listener",
 ] }
-metrics-util = "0.16.0"
+metrics-util = "0.17.0"
 mime = "0.3.16"
 minify-html = "0.15.0"
-opentelemetry = "0.22"
-opentelemetry_sdk = { version = "0.22", features = ["rt-tokio"] }
-opentelemetry-otlp = "0.15"
+opentelemetry = "0.27.1"
+opentelemetry_sdk = { version = "0.27", features = ["rt-tokio"] }
+opentelemetry-otlp = { version = "0.27", features = ["grpc-tonic"] }
 pin-project-lite = "0.2.9"
 # pinned to metrics-util
 quanta = "0.12.0"
 rand = "0.8"
-reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "stream"]}
-reqwest-middleware = "0.2"
-reqwest-tracing = "0.4.5"
+reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream"]}
+reqwest-middleware = { version = "0.4", default-features = false, features = ["json"] }
+reqwest-tracing = "0.5.0"
 ring = "0.17.5"
 rsa = "0.9"
 rsa-magic-public-key = "0.8.0"
-rustls = "0.22.0"
-rustls-channel-resolver = "0.2.0"
+rustls = { version = "0.23.0", default-features = false, features = ["ring", "logging", "std", "tls12"] }
+rustls-channel-resolver = "0.3.0"
 rustls-pemfile = "2"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sled = "0.34.7"
 streem = "0.2.0"
-teloxide = { version = "0.12.0", default-features = false, features = [
+teloxide = { version = "0.13.0", default-features = false, features = [
   "ctrlc_handler",
   "macros",
   "rustls",
 ] }
-thiserror = "1.0"
+thiserror = "2.0"
 time = { version = "0.3.17", features = ["serde"] }
 tracing = "0.1"
 tracing-error = "0.2"
 tracing-log = "0.2"
-tracing-opentelemetry = "0.23"
+tracing-opentelemetry = "0.28"
 tracing-subscriber = { version = "0.3", features = [
   "ansi",
   "env-filter",
@@ -86,17 +87,17 @@ tokio = { version = "1", features = ["full", "tracing"] }
 uuid = { version = "1", features = ["v4", "serde"] }
 
 [dependencies.background-jobs]
-version = "0.18.0"
+version = "0.19.0"
 default-features = false
 features = ["error-logging", "metrics", "tokio"]
 
 [dependencies.http-signature-normalization-actix]
-version = "0.11.0"
+version = "0.11.1"
 default-features = false
 features = ["server", "ring"]
 
 [dependencies.http-signature-normalization-reqwest]
-version = "0.11.0"
+version = "0.13.0"
 default-features = false
 features = ["middleware", "ring"]
 

docker/prod/docker-compose.yml

@@ -2,7 +2,7 @@ version: '3.3'
 
 services:
   relay:
-    image: asonix/relay:0.3.108
+    image: asonix/relay:0.3.115
     ports:
       - "8079:8079"
     restart: always

flake.lock

@@ -20,16 +20,16 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1711163522,
-        "narHash": "sha256-YN/Ciidm+A0fmJPWlHBGvVkcarYWSC+s3NTPk/P+q3c=",
+        "lastModified": 1733550349,
+        "narHash": "sha256-NcGumB4Lr6KSDq+nIqXtNA8QwAQKDSZT7N9OTGWbTrs=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "44d0940ea560dee511026a53f0e2e2cde489b4d4",
+        "rev": "e2605d0744c2417b09f8bf850dfca42fcf537d34",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "nixos-unstable",
+        "ref": "nixos-24.11",
         "repo": "nixpkgs",
         "type": "github"
       }

flake.nix

@@ -2,7 +2,7 @@
   description = "relay";
 
   inputs = {
-    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-24.11";
     flake-utils.url = "github:numtide/flake-utils";
   };
 

relay.nix

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

src/admin/client.rs

@@ -5,7 +5,6 @@ use crate::{
     error::{Error, ErrorKind},
     extractors::XApiToken,
 };
-use actix_web::http::header::Header;
 use reqwest_middleware::ClientWithMiddleware;
 use serde::de::DeserializeOwned;
 
@@ -87,13 +86,17 @@ async fn get_results<T: DeserializeOwned>(
 
     let res = client
         .get(iri.as_str())
-        .header(XApiToken::name(), x_api_token.to_string())
+        .header(XApiToken::http1_name(), x_api_token.to_string())
         .send()
         .await
         .map_err(|e| ErrorKind::SendRequest(iri.to_string(), e.to_string()))?;
 
     if !res.status().is_success() {
-        return Err(ErrorKind::Status(iri.to_string(), res.status()).into());
+        return Err(ErrorKind::Status(
+            iri.to_string(),
+            crate::http1::status_to_http02(res.status()),
+        )
+        .into());
     }
 
     let t = res
@@ -116,7 +119,7 @@ async fn post_domains(
 
     let res = client
         .post(iri.as_str())
-        .header(XApiToken::name(), x_api_token.to_string())
+        .header(XApiToken::http1_name(), x_api_token.to_string())
         .json(&Domains { domains })
         .send()
         .await

src/db.rs

@@ -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());
                 }
 

src/error.rs

@@ -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),
 

src/extractors.rs

@@ -163,6 +163,10 @@ impl XApiToken {
     pub(crate) fn new(token: String) -> Self {
         Self(token)
     }
+
+    pub(crate) const fn http1_name() -> reqwest::header::HeaderName {
+        reqwest::header::HeaderName::from_static("x-api-token")
+    }
 }
 
 impl Header for XApiToken {

src/http1.rs

@@ -0,0 +1,18 @@
+pub(crate) fn name_to_http02(
+    name: &reqwest::header::HeaderName,
+) -> actix_web::http::header::HeaderName {
+    actix_web::http::header::HeaderName::from_bytes(name.as_ref())
+        .expect("headername conversions always work")
+}
+
+pub(crate) fn value_to_http02(
+    value: &reqwest::header::HeaderValue,
+) -> actix_web::http::header::HeaderValue {
+    actix_web::http::header::HeaderValue::from_bytes(value.as_bytes())
+        .expect("headervalue conversions always work")
+}
+
+pub(crate) fn status_to_http02(status: reqwest::StatusCode) -> actix_web::http::StatusCode {
+    actix_web::http::StatusCode::from_u16(status.as_u16())
+        .expect("statuscode conversions always work")
+}

src/jobs/nodeinfo.rs

@@ -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.";

src/main.rs

@@ -12,7 +12,7 @@ use error::Error;
 use http_signature_normalization_actix::middleware::VerifySignature;
 use metrics_exporter_prometheus::PrometheusBuilder;
 use metrics_util::layers::FanoutBuilder;
-use opentelemetry::KeyValue;
+use opentelemetry::{trace::TracerProvider, KeyValue};
 use opentelemetry_otlp::WithExportConfig;
 use opentelemetry_sdk::Resource;
 use reqwest_middleware::ClientWithMiddleware;
@@ -33,11 +33,13 @@ mod db;
 mod error;
 mod extractors;
 mod future;
+mod http1;
 mod jobs;
 mod middleware;
 mod requests;
 mod routes;
 mod spawner;
+mod stream;
 mod telegram;
 
 use crate::config::UrlKind;
@@ -81,22 +83,21 @@ fn init_subscriber(
     let subscriber = subscriber.with(console_layer);
 
     if let Some(url) = opentelemetry_url {
-        let tracer = opentelemetry_otlp::new_pipeline()
-            .tracing()
-            .with_trace_config(
-                opentelemetry_sdk::trace::config().with_resource(Resource::new(vec![
-                    KeyValue::new("service.name", software_name),
-                ])),
-            )
-            .with_exporter(
-                opentelemetry_otlp::new_exporter()
-                    .tonic()
-                    .with_endpoint(url.as_str()),
-            )
-            .install_batch(opentelemetry_sdk::runtime::Tokio)?;
+        let exporter = opentelemetry_otlp::SpanExporter::builder()
+            .with_tonic()
+            .with_endpoint(url.as_str())
+            .build()?;
+
+        let tracer_provider = opentelemetry_sdk::trace::TracerProvider::builder()
+            .with_resource(Resource::new(vec![KeyValue::new(
+                "service.name",
+                software_name,
+            )]))
+            .with_batch_exporter(exporter, opentelemetry_sdk::runtime::Tokio)
+            .build();
 
         let otel_layer = tracing_opentelemetry::layer()
-            .with_tracer(tracer)
+            .with_tracer(tracer_provider.tracer(software_name))
             .with_filter(targets);
 
         let subscriber = subscriber.with(otel_layer);
@@ -417,7 +418,7 @@ async fn server_main(
             .with_no_client_auth()
             .with_cert_resolver(cert_rx);
         server
-            .bind_rustls_0_22(bind_address, server_config)?
+            .bind_rustls_0_23(bind_address, server_config)?
             .run()
             .await?;
 

src/requests.rs

@@ -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
@@ -229,7 +233,11 @@ impl Requests {
                 }
             }
 
-            return Err(ErrorKind::Status(parsed_url.to_string(), status).into());
+            return Err(ErrorKind::Status(
+                parsed_url.to_string(),
+                crate::http1::status_to_http02(status),
+            )
+            .into());
         }
 
         // only actually succeed a breaker on 2xx response
@@ -262,7 +270,7 @@ impl Requests {
     where
         T: serde::de::DeserializeOwned,
     {
-        let body = self
+        let stream = self
             .do_deliver(
                 url,
                 &serde_json::json!({}),
@@ -271,8 +279,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 +308,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)?)
     }

src/routes/media.rs

@@ -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>,
@@ -19,13 +23,19 @@ pub(crate) async fn route(
             .fetch_response(&url, BreakerStrategy::Allow404AndBelow)
             .await?;
 
-        let mut response = HttpResponse::build(res.status());
+        let mut response = HttpResponse::build(crate::http1::status_to_http02(res.status()));
 
         for (name, value) in res.headers().iter().filter(|(h, _)| *h != "connection") {
-            response.insert_header((name.clone(), value.clone()));
+            response.insert_header((
+                crate::http1::name_to_http02(name),
+                crate::http1::value_to_http02(value),
+            ));
         }
 
-        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())

src/stream.rs

@@ -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())
+}

src/telegram.rs

@@ -75,7 +75,8 @@ pub(crate) fn start(admin_handle: String, db: Db, token: &str) {
 
 fn is_admin(admin_handle: &str, message: &Message) -> bool {
     message
-        .from()
+        .from
+        .as_ref()
         .and_then(|user| user.username.as_deref())
         .map(|username| username == admin_handle)
         .unwrap_or(false)