astro/caveman
astro
/
caveman
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

buzzback: init

This commit is contained in:
Astro 2023-10-12 20:11:21 +02:00
parent 9378cfc0b3
commit 42728d0efb
11 changed files with 482 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
Cargo.lock generated
View File

@ -253,6 +253,18 @@ dependencies = [
"rustc-demangle",
]
[[package]]
name = "base64"
version = "0.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8"
[[package]]
name = "base64"
version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ea22880d78093b0cbe17c89f64a7d457941e65759157ec6cb31a31d652b05e5"
[[package]]
name = "base64"
version = "0.21.4"
@ -316,6 +328,26 @@ version = "3.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f30e7476521f6f8af1a1c4c0b8cc94f0bee37d91763d0ca2665f299b6cd8aec"
[[package]]
name = "buzzback"
version = "0.1.0"
dependencies = [
"http",
"http_digest_headers",
"httpdate",
"reqwest",
"serde",
"serde_json",
"serde_yaml",
"sigh",
"thiserror",
"tokio",
"tokio-postgres",
"tracing",
"tracing-subscriber",
"urlencoding 2.1.3",
]
[[package]]
name = "byteorder"
version = "1.5.0"
@ -391,7 +423,7 @@ dependencies = [
"serde_yaml",
"tokio",
"tracing",
"urlencoding",
"urlencoding 1.3.3",
]
[[package]]
@ -413,7 +445,7 @@ dependencies = [
"tokio",
"tokio-uring",
"tracing",
"urlencoding",
"urlencoding 1.3.3",
]
[[package]]
@ -914,6 +946,18 @@ version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "add0ab9360ddbd88cfeb3bd9574a1d85cfdfa14db10b3e21d3700dbc4328758f"
[[package]]
name = "http_digest_headers"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7db39af769c6cb000b7ea765bd5eedcc812d3ec00f3ea50e073c19e00a75916d"
dependencies = [
"anyhow",
"base64 0.13.1",
"openssl",
"thiserror",
]
[[package]]
name = "httparse"
version = "1.8.0"
@ -1571,7 +1615,7 @@ version = "0.6.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49b6c5ef183cd3ab4ba005f1ca64c21e8bd97ce4699cfea9e8d9a2c4958ca520"
dependencies = [
"base64",
"base64 0.21.4",
"byteorder",
"bytes",
"fallible-iterator",
@ -1771,7 +1815,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "046cd98826c46c2ac8ddecae268eb5c2e58628688a5fc7a2643704a73faba95b"
dependencies = [
"async-compression",
"base64",
"base64 0.21.4",
"bytes",
"encoding_rs",
"futures-core",
@ -1967,6 +2011,19 @@ dependencies = [
"lazy_static",
]
[[package]]
name = "sigh"
version = "1.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "17a67f3294aadf0022646d41c33b888d79db23670e4c34b1bc007c6f901a3b77"
dependencies = [
"base64 0.20.0",
"http",
"nom",
"openssl",
"thiserror",
]
[[package]]
name = "signal-hook-registry"
version = "1.4.1"
@ -2509,6 +2566,12 @@ version = "1.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a1f0175e03a0973cf4afd476bef05c26e228520400eb1fd473ad417b1c00ffb"
[[package]]
name = "urlencoding"
version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
[[package]]
name = "utf8-cstr"
version = "0.1.6"

1
Cargo.toml
View File

@ -5,6 +5,7 @@ members = [
"butcher",
"gatherer",
"smokestack",
"buzzback",
]
resolver = "2"

21
buzzback/Cargo.toml Normal file
View File

@ -0,0 +1,21 @@
[package]
name = "buzzback"
description = "Follows back"
version = "0.1.0"
edition = "2021"
[dependencies]
tokio = { version = "1", features = ["full", "time"] }
tracing = "*"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
serde = { version = "1", features = ["serde_derive"] }
serde_json = "1"
serde_yaml = "0.9"
reqwest = { version = "0.11", features = ["json", "stream"] }
sigh = "1.0"
http_digest_headers = { version="0.1.0", default-features = false, features = ["use_openssl"] }
thiserror = "1"
http = "0.2"
tokio-postgres = "0.7"
urlencoding = "2"
httpdate = "1"

45
buzzback/config.yaml Normal file
View File

@ -0,0 +1,45 @@
# external https hostname
hostname: relay.fedi.buzz
# ActivityPub signing keypair
priv_key_file: private-key.pem
#pub_key_file: public-key.pem
# PostgreSQL
db: "host=localhost user=relay password=xyz dbname=buzzrelay"
default_actor: https://relay.fedi.buzz/tag/FediBuzz
relays:
- https://relay.c.im/inbox
- https://relay.beckmeyer.us/inbox
- https://relay.froth.zone/inbox
- https://relay.intahnet.co.uk/inbox
- https://relay.infosec.exchange/inbox
- https://relay.toot.io/inbox
- https://bigrelay.social/inbox
- https://relay.toot.yukimochi.jp/inbox
- https://relay.an.exchange/inbox
- https://relay.fedinet.social/inbox
- https://relay.intahnet.co.uk/inbox
- https://en.relay.friendi.ca/inbox
- https://relay.101010.pl/inbox
- https://relay.chocoflan.net/inbox
- https://relay.dresden.network/inbox
- https://mastodon-relay.moew.science/inbox
- https://relay.beckmeyer.us/inbox
- https://relay.fedi.agency/inbox
- https://relay.fedinet.social/inbox
- https://relay.flm9.me/inbox
- https://relay.froth.zone/inbox
- https://relay.homunyan.com/inbox
- https://relay.intahnet.co.uk/inbox
- https://relay.libranet.de/inbox
- https://relay.masto.la/inbox
- https://relay.minecloud.ro/inbox
- https://relay.mistli.net/inbox
- https://relay.pissdichal.de/inbox
- https://relay.toot.yukimochi.jp/inbox
- https://relay.uggs.io/inbox
- https://relay.wagnersnetz.de/inbox
- https://relay.wig.gl/inbox
- https://relay.dog/inbox
- https://relay.darmstadt.social/inbox
- https://relay.douzepoints.social/inbox

48
buzzback/src/activitypub.rs Normal file
View File

@ -0,0 +1,48 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Actor {
#[serde(rename = "@context")]
pub jsonld_context: serde_json::Value,
#[serde(rename = "type")]
pub actor_type: String,
pub id: String,
pub name: Option<String>,
pub icon: Option<Media>,
pub inbox: String,
pub outbox: String,
#[serde(rename = "publicKey")]
pub public_key: ActorPublicKey,
#[serde(rename = "preferredUsername")]
pub preferred_username: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ActorPublicKey {
pub id: String,
pub owner: Option<String>,
#[serde(rename = "publicKeyPem")]
pub pem: String,
}
/// `ActivityPub` "activity"
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Action<O> {
#[serde(rename = "@context")]
pub jsonld_context: serde_json::Value,
#[serde(rename = "type")]
pub action_type: String,
pub id: String,
pub actor: String,
pub to: Option<serde_json::Value>,
pub object: Option<O>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Media {
#[serde(rename = "type")]
pub media_type: String,
#[serde(rename = "mediaType")]
pub content_type: String,
pub url: String,
}

35
buzzback/src/config.rs Normal file
View File

@ -0,0 +1,35 @@
use serde::Deserialize;
use sigh::{PrivateKey, Key};
#[derive(Deserialize)]
pub struct Config {
pub db: String,
pub hostname: String,
priv_key_file: String,
// pub_key_file: String,
pub default_actor: String,
pub relays: Vec<String>,
}
impl Config {
pub fn load(config_file: &str) -> Config {
let data = std::fs::read_to_string(config_file)
.expect("read config");
serde_yaml::from_str(&data)
.expect("parse config")
}
pub fn priv_key(&self) -> PrivateKey {
let data = std::fs::read_to_string(&self.priv_key_file)
.expect("read priv_key_file");
PrivateKey::from_pem(data.as_bytes())
.expect("priv_key")
}
// pub fn pub_key(&self) -> PublicKey {
// let data = std::fs::read_to_string(&self.pub_key_file)
// .expect("read pub_key_file");
// PublicKey::from_pem(data.as_bytes())
// .expect("pub_key")
// }
}

54
buzzback/src/db.rs Normal file
View File

@ -0,0 +1,54 @@
use std::sync::Arc;
use tokio_postgres::{Client, Error, NoTls};
#[derive(Clone, Debug)]
pub struct Follow {
/// Follower identification
pub id: String,
/// Subscribing inbox that is supposed to receive pushes
pub inbox: String,
/// Followed actor
pub actor: String,
}
#[derive(Clone)]
pub struct Database {
inner: Arc<DatabaseInner>,
}
struct DatabaseInner {
client: Client,
}
impl Database {
pub async fn connect(conn_str: &str) -> Self {
let (client, connection) = tokio_postgres::connect(conn_str, NoTls)
.await
.unwrap();
tokio::spawn(async move {
if let Err(e) = connection.await {
tracing::error!("postgresql: {}", e);
}
});
Database {
inner: Arc::new(DatabaseInner {
client,
}),
}
}
pub async fn get_instance_followers(&self) -> Result<impl Iterator<Item = Follow>, Error> {
let rows = self.inner.client.query("SELECT inbox, MIN(id), MIN(actor) FROM follows WHERE inbox LIKE '%/actor/inbox' GROUP BY inbox", &[])
.await?;
Ok(rows.into_iter()
.map(|row| Follow {
id: row.get(1),
inbox: row.get(0),
actor: row.get(2),
})
)
}
}

20
buzzback/src/digest.rs Normal file
View File

@ -0,0 +1,20 @@
use http_digest_headers::{DigestHeader, DigestMethod};
pub fn generate_header(body: &[u8]) -> Result<String, ()> {
let mut digest_header = DigestHeader::new()
.with_method(DigestMethod::SHA256, body)
.map(|h| format!("{}", h))
.map_err(|_| ())?;
// mastodon expects uppercase algo name
if digest_header.starts_with("sha-") {
digest_header.replace_range(..4, "SHA-");
}
// mastodon uses base64::alphabet::STANDARD, not base64::alphabet::URL_SAFE
digest_header.replace_range(
7..,
&digest_header[7..].replace('-', "+").replace('_', "/")
);
Ok(digest_header)
}

17
buzzback/src/error.rs Normal file
View File

@ -0,0 +1,17 @@
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("HTTP Digest generation error")]
Digest,
#[error("JSON encoding error")]
Json(#[from] serde_json::Error),
#[error("Signature error")]
Signature(#[from] sigh::Error),
#[error("HTTP request error")]
HttpReq(#[from] http::Error),
#[error("HTTP client error")]
Http(#[from] reqwest::Error),
#[error("Invalid URI")]
InvalidUri,
#[error("Error response from remote")]
Response(String),
}

117
buzzback/src/main.rs Normal file
View File

@ -0,0 +1,117 @@
use sigh::PrivateKey;
use std::{sync::Arc, time::Duration};
use std::{panic, process};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
mod error;
mod config;
mod db;
mod activitypub;
mod digest;
mod send;
async fn follow_back(
client: &reqwest::Client, hostname: &str,
priv_key: &PrivateKey,
follow: db::Follow,
) {
let follow_id = format!(
"https://{}/activity/follow/{}/{}",
hostname,
urlencoding::encode(&follow.id),
urlencoding::encode(&follow.actor),
);
let action = activitypub::Action {
jsonld_context: serde_json::Value::String("https://www.w3.org/ns/activitystreams".to_string()),
action_type: "Follow".to_string(),
id: follow_id,
to: None,
actor: follow.actor.clone(),
object: Some(&follow.id),
};
let key_id = format!("{}#key", follow.actor);
let result = send::send(
client, &follow.inbox,
&key_id, &priv_key,
&action,
).await;
match result {
Ok(()) => {
tracing::info!("Ok {}", follow.id);
}
Err(e) => {
tracing::error!("POST: {:?}", e);
}
}
}
#[tokio::main]
async fn main() {
exit_on_panic();
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| {
"buzzback=trace,tower_http=trace,axum=trace".into()
}),
)
.with(tracing_subscriber::fmt::layer())
.init();
let config = config::Config::load(
&std::env::args().nth(1)
.expect("Call with config.yaml")
);
let priv_key = config.priv_key();
let client = Arc::new(
reqwest::Client::builder()
.timeout(Duration::from_secs(5))
.user_agent(concat!(
env!("CARGO_PKG_NAME"),
"/",
env!("CARGO_PKG_VERSION"),
))
.pool_max_idle_per_host(0)
.pool_idle_timeout(Some(Duration::from_secs(1)))
.build()
.unwrap()
);
// Follow relays from the static list in the config.yaml
for relay in config.relays {
let (id, inbox) =
if relay.ends_with("/inbox") {
(relay.replace("/inbox", "/actor"), relay.clone())
} else if relay.ends_with("/actor") {
(relay.clone(), relay.replace("/actor", "/inbox"))
} else {
panic!("Not sure how to deal with relay {}", relay);
};
tracing::trace!("Following {}", &id);
let follow = db::Follow {
id,
actor: config.default_actor.clone(),
inbox,
};
follow_back(&client, &config.hostname, &priv_key, follow).await;
}
let database = db::Database::connect(&config.db).await;
let instance_followers = database.get_instance_followers().await.unwrap();
// Follow back every instance that followed us
for follow in instance_followers {
tracing::trace!("Following {}", follow.id);
follow_back(&client, &config.hostname, &priv_key, follow).await;
}
}
fn exit_on_panic() {
let orig_hook = panic::take_hook();
panic::set_hook(Box::new(move |panic_info| {
// invoke the default handler and exit the process
orig_hook(panic_info);
process::exit(1);
}));
}

57
buzzback/src/send.rs Normal file
View File

@ -0,0 +1,57 @@
use std::{
sync::Arc,
time::SystemTime,
};
use http::StatusCode;
use serde::Serialize;
use sigh::{PrivateKey, SigningConfig, alg::RsaSha256};
use crate::{digest, error::Error};
pub async fn send<T: Serialize>(
client: &reqwest::Client,
uri: &str,
key_id: &str,
private_key: &PrivateKey,
body: &T,
) -> Result<(), Error> {
let body = Arc::new(
serde_json::to_vec(body)
.map_err(Error::Json)?
);
send_raw(client, uri, key_id, private_key, body).await
}
pub async fn send_raw(
client: &reqwest::Client,
uri: &str,
key_id: &str,
private_key: &PrivateKey,
body: Arc<Vec<u8>>,
) -> Result<(), Error> {
let url = reqwest::Url::parse(uri)
.map_err(|_| Error::InvalidUri)?;
let host = format!("{}", url.host().ok_or(Error::InvalidUri)?);
let digest_header = digest::generate_header(&body)
.map_err(|()| Error::Digest)?;
let mut req = http::Request::builder()
.method("POST")
.uri(uri)
.header("host", &host)
.header("content-type", "application/activity+json")
.header("date", httpdate::fmt_http_date(SystemTime::now()))
.header("digest", digest_header)
.body(body.as_ref().clone())
.map_err(Error::HttpReq)?;
SigningConfig::new(RsaSha256, private_key, key_id)
.sign(&mut req)?;
let req: reqwest::Request = req.try_into()?;
let res = client.execute(req)
.await?;
if res.status() >= StatusCode::OK && res.status() < StatusCode::MULTIPLE_CHOICES {
Ok(())
} else {
tracing::error!("send_raw {} response HTTP {}", url, res.status());
let response = res.text().await?;
Err(Error::Response(response))
}
}