rafaelcaricio/relay
1
0
Fork 0
forked from mirrors/relay
Code Pull requests Activity

Update to latest activitystreams

Browse source
This commit is contained in:
Aode (Lion) 2022-01-17 16:54:45 -06:00
parent 6b0d3298cc
commit 8893895c71
27 changed files with 338 additions and 264 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
Cargo.lock generated
View file

@ -4,17 +4,16 @@ version = 3
[[package]]
name = "activitystreams"
version = "0.7.0-alpha.14"
version = "0.7.0-alpha.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6bcc3fbb392890a1942b1e5cca76cba93c8ed24b5ff50004cc3289afaab3f92c"
checksum = "31bca51dcfddda6551570371a1c065528050eea2151a29897ac2dad67b5624cb"
dependencies = [
"activitystreams-kinds",
"chrono",
"iri-string",
"mime",
"serde 1.0.133",
"serde_json",
"thiserror",
"url",
"time 0.3.5",
]
[[package]]
@ -30,12 +29,12 @@ dependencies = [
[[package]]
name = "activitystreams-kinds"
version = "0.1.2"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0784e99afd032199d3ed70cefb8eb3a8d1aef15f7f2c4e68d033c4e12bb6079e"
checksum = "4a762e3441050b51cd0695c1413735cd04195950f50dba8f5da5d7201628fcfc"
dependencies = [
"iri-string",
"serde 1.0.133",
"url",
]
[[package]]
@ -318,7 +317,6 @@ dependencies = [
"awc",
"background-jobs",
"base64",
"chrono",
"config",
"console-subscriber",
"dashmap",
@ -339,6 +337,7 @@ dependencies = [
"sled",
"structopt",
"thiserror",
"time 0.3.5",
"toml",
"tracing",
"tracing-actix-web",
@ -1256,6 +1255,15 @@ dependencies = [
"cfg-if",
]
[[package]]
name = "iri-string"
version = "0.5.0-beta.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "08b57553d1be311c9e979117451952d55f1b33366abe5a07f07899dc2f11a4d9"
dependencies = [
"serde 1.0.133",
]
[[package]]
name = "itertools"
version = "0.10.3"
@ -3045,7 +3053,6 @@ dependencies = [
"idna",
"matches",
"percent-encoding",
"serde 1.0.133",
]
[[package]]

4
Cargo.toml
View file

@ -21,13 +21,12 @@ anyhow = "1.0"
actix-rt = "2.0.2"
actix-web = { version = "4.0.0-beta.7", default-features = false }
actix-webfinger = "0.4.0-beta.5"
activitystreams = "0.7.0-alpha.10"
activitystreams = "0.7.0-alpha.16"
activitystreams-ext = "0.1.0-alpha.2"
ammonia = "3.1.0"
async-rwlock = "1.3.0"
awc = { version = "3.0.0-beta.6", default-features = false, features = ["rustls"] }
base64 = "0.13"
chrono = "0.4.19"
config = "0.11.0"
console-subscriber = "0.1"
dashmap = "5.0.0"
@ -46,6 +45,7 @@ sha2 = "0.10"
sled = "0.34.6"
structopt = "0.3.12"
thiserror = "1.0"
time = "0.3.5"
tracing = "0.1"
tracing-awc = { version = "0.1.0-beta.10" }
tracing-error = "0.2"

6
src/apub.rs
View file

@ -2,15 +2,15 @@ use activitystreams::{
activity::ActorAndObject,
actor::{Actor, ApActor},
unparsed::UnparsedMutExt,
url::Url,
iri_string::types::IriString,
};
use activitystreams_ext::{Ext1, UnparsedExtension};
#[derive(Clone, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PublicKeyInner {
pub id: Url,
pub owner: Url,
pub id: IriString,
pub owner: IriString,
pub public_key_pem: String,
}

72
src/config.rs
View file

@ -4,7 +4,13 @@ use crate::{
middleware::MyVerify,
requests::Requests,
};
use activitystreams::{uri, url::Url};
use activitystreams::{
iri,
iri_string::{
resolve::FixedBaseResolver,
types::{IriAbsoluteString, IriFragmentStr, IriRelativeStr, IriString},
},
};
use config::Environment;
use http_signature_normalization_actix::prelude::{VerifyDigest, VerifySignature};
use sha2::{Digest, Sha256};
@ -22,8 +28,8 @@ pub(crate) struct ParsedConfig {
https: bool,
publish_blocks: bool,
sled_path: PathBuf,
source_repo: Url,
opentelemetry_url: Option<Url>,
source_repo: IriString,
opentelemetry_url: Option<IriString>,
}
#[derive(Clone)]
@ -35,12 +41,13 @@ pub struct Config {
restricted_mode: bool,
validate_signatures: bool,
publish_blocks: bool,
base_uri: Url,
base_uri: IriAbsoluteString,
sled_path: PathBuf,
source_repo: Url,
opentelemetry_url: Option<Url>,
source_repo: IriString,
opentelemetry_url: Option<IriString>,
}
#[derive(Debug)]
pub enum UrlKind {
Activity,
Actor,
@ -95,7 +102,7 @@ impl Config {
let config: ParsedConfig = config.try_into()?;
let scheme = if config.https { "https" } else { "http" };
let base_uri = uri!(format!("{}://{}", scheme, config.hostname));
let base_uri = iri!(format!("{}://{}", scheme, config.hostname)).into_absolute();
Ok(Config {
hostname: config.hostname,
@ -210,33 +217,50 @@ impl Config {
)
}
pub(crate) fn source_code(&self) -> &Url {
pub(crate) fn source_code(&self) -> &IriString {
&self.source_repo
}
pub(crate) fn opentelemetry_url(&self) -> Option<&Url> {
pub(crate) fn opentelemetry_url(&self) -> Option<&IriString> {
self.opentelemetry_url.as_ref()
}
pub(crate) fn generate_url(&self, kind: UrlKind) -> Url {
let mut url = self.base_uri.clone();
pub(crate) fn generate_url(&self, kind: UrlKind) -> IriString {
self.do_generate_url(kind).expect("Generated valid IRI")
}
match kind {
UrlKind::Activity => url.set_path(&format!("activity/{}", Uuid::new_v4())),
UrlKind::Actor => url.set_path("actor"),
UrlKind::Followers => url.set_path("followers"),
UrlKind::Following => url.set_path("following"),
UrlKind::Inbox => url.set_path("inbox"),
UrlKind::Index => (),
#[tracing::instrument(fields(base_uri = tracing::field::debug(&self.base_uri), kind = tracing::field::debug(&kind)))]
fn do_generate_url(&self, kind: UrlKind) -> Result<IriString, Error> {
let iri = match kind {
UrlKind::Activity => FixedBaseResolver::new(self.base_uri.as_ref())
.resolve(IriRelativeStr::new(&format!("activity/{}", Uuid::new_v4()))?.as_ref())?,
UrlKind::Actor => FixedBaseResolver::new(self.base_uri.as_ref())
.resolve(IriRelativeStr::new("actor")?.as_ref())?,
UrlKind::Followers => FixedBaseResolver::new(self.base_uri.as_ref())
.resolve(IriRelativeStr::new("followers")?.as_ref())?,
UrlKind::Following => FixedBaseResolver::new(self.base_uri.as_ref())
.resolve(IriRelativeStr::new("following")?.as_ref())?,
UrlKind::Inbox => FixedBaseResolver::new(self.base_uri.as_ref())
.resolve(IriRelativeStr::new("inbox")?.as_ref())?,
UrlKind::Index => self.base_uri.clone().into(),
UrlKind::MainKey => {
url.set_path("actor");
url.set_fragment(Some("main-key"));
let actor = IriRelativeStr::new("actor")?;
let fragment = IriFragmentStr::new("main-key")?;
let mut resolved =
FixedBaseResolver::new(self.base_uri.as_ref()).resolve(actor.as_ref())?;
resolved.set_fragment(Some(fragment));
resolved
}
UrlKind::Media(uuid) => url.set_path(&format!("media/{}", uuid)),
UrlKind::NodeInfo => url.set_path("nodeinfo/2.0.json"),
UrlKind::Outbox => url.set_path("outbox"),
UrlKind::Media(uuid) => FixedBaseResolver::new(self.base_uri.as_ref())
.resolve(IriRelativeStr::new(&format!("media/{}", uuid))?.as_ref())?,
UrlKind::NodeInfo => FixedBaseResolver::new(self.base_uri.as_ref())
.resolve(IriRelativeStr::new("nodeinfo/2.0.json")?.as_ref())?,
UrlKind::Outbox => FixedBaseResolver::new(self.base_uri.as_ref())
.resolve(IriRelativeStr::new("outbox")?.as_ref())?,
};
url
Ok(iri)
}
}

16
src/data/actor.rs
View file

@ -4,7 +4,7 @@ use crate::{
error::{Error, ErrorKind},
requests::Requests,
};
use activitystreams::{prelude::*, url::Url};
use activitystreams::{iri_string::types::IriString, prelude::*};
use std::time::{Duration, SystemTime};
const REFETCH_DURATION: Duration = Duration::from_secs(60 * 30);
@ -40,7 +40,7 @@ impl ActorCache {
#[tracing::instrument(name = "Get Actor", fields(id = id.to_string().as_str(), requests))]
pub(crate) async fn get(
&self,
id: &Url,
id: &IriString,
requests: &Requests,
) -> Result<MaybeCached<Actor>, Error> {
if let Some(actor) = self.db.actor(id.clone()).await? {
@ -66,12 +66,16 @@ impl ActorCache {
}
#[tracing::instrument(name = "Fetch remote actor", fields(id = id.to_string().as_str(), requests))]
pub(crate) async fn get_no_cache(&self, id: &Url, requests: &Requests) -> Result<Actor, Error> {
pub(crate) async fn get_no_cache(
&self,
id: &IriString,
requests: &Requests,
) -> Result<Actor, Error> {
let accepted_actor = requests.fetch::<AcceptedActors>(id.as_str()).await?;
let input_domain = id.domain().ok_or(ErrorKind::MissingDomain)?;
let input_authority = id.authority_components().ok_or(ErrorKind::MissingDomain)?;
let accepted_actor_id = accepted_actor
.id(input_domain)?
.id(input_authority.host(), input_authority.port())?
.ok_or(ErrorKind::MissingId)?;
let inbox = get_inbox(&accepted_actor)?.clone();
@ -90,7 +94,7 @@ impl ActorCache {
}
}
fn get_inbox(actor: &AcceptedActors) -> Result<&Url, Error> {
fn get_inbox(actor: &AcceptedActors) -> Result<&IriString, Error> {
Ok(actor
.endpoints()?
.and_then(|e| e.shared_inbox)

8
src/data/media.rs
View file

@ -2,7 +2,7 @@ use crate::{
db::{Db, MediaMeta},
error::Error,
};
use activitystreams::url::Url;
use activitystreams::iri_string::types::IriString;
use actix_web::web::Bytes;
use std::time::{Duration, SystemTime};
use uuid::Uuid;
@ -20,12 +20,12 @@ impl MediaCache {
}
#[tracing::instrument(name = "Get media uuid", fields(url = url.to_string().as_str()))]
pub(crate) async fn get_uuid(&self, url: Url) -> Result<Option<Uuid>, Error> {
pub(crate) async fn get_uuid(&self, url: IriString) -> Result<Option<Uuid>, Error> {
self.db.media_id(url).await
}
#[tracing::instrument(name = "Get media url")]
pub(crate) async fn get_url(&self, uuid: Uuid) -> Result<Option<Url>, Error> {
pub(crate) async fn get_url(&self, uuid: Uuid) -> Result<Option<IriString>, Error> {
self.db.media_url(uuid).await
}
@ -56,7 +56,7 @@ impl MediaCache {
}
#[tracing::instrument(name = "Store media url", fields(url = url.to_string().as_str()))]
pub(crate) async fn store_url(&self, url: Url) -> Result<Uuid, Error> {
pub(crate) async fn store_url(&self, url: IriString) -> Result<Uuid, Error> {
let uuid = Uuid::new_v4();
self.db.save_url(url, uuid).await?;

44
src/data/node.rs
View file

@ -1,8 +1,8 @@
use crate::{
db::{Contact, Db, Info, Instance},
error::Error,
error::{Error, ErrorKind},
};
use activitystreams::url::Url;
use activitystreams::{iri, iri_string::types::IriString};
use std::time::{Duration, SystemTime};
#[derive(Clone, Debug)]
@ -12,7 +12,7 @@ pub struct NodeCache {
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub struct Node {
pub(crate) base: Url,
pub(crate) base: IriString,
pub(crate) info: Option<Info>,
pub(crate) instance: Option<Instance>,
pub(crate) contact: Option<Contact>,
@ -50,18 +50,15 @@ impl NodeCache {
let instance = instances.get(&actor_id).cloned();
let contact = contacts.get(&actor_id).cloned();
Node::new(actor_id)
.info(info)
.instance(instance)
.contact(contact)
Node::new(actor_id).map(|node| node.info(info).instance(instance).contact(contact))
})
.collect();
.collect::<Result<Vec<Node>, Error>>()?;
Ok(vec)
}
#[tracing::instrument(name = "Is NodeInfo Outdated", fields(actor_id = actor_id.to_string().as_str()))]
pub(crate) async fn is_nodeinfo_outdated(&self, actor_id: Url) -> bool {
pub(crate) async fn is_nodeinfo_outdated(&self, actor_id: IriString) -> bool {
self.db
.info(actor_id)
.await
@ -70,7 +67,7 @@ impl NodeCache {
}
#[tracing::instrument(name = "Is Contact Outdated", fields(actor_id = actor_id.to_string().as_str()))]
pub(crate) async fn is_contact_outdated(&self, actor_id: Url) -> bool {
pub(crate) async fn is_contact_outdated(&self, actor_id: IriString) -> bool {
self.db
.contact(actor_id)
.await
@ -79,7 +76,7 @@ impl NodeCache {
}
#[tracing::instrument(name = "Is Instance Outdated", fields(actor_id = actor_id.to_string().as_str()))]
pub(crate) async fn is_instance_outdated(&self, actor_id: Url) -> bool {
pub(crate) async fn is_instance_outdated(&self, actor_id: IriString) -> bool {
self.db
.instance(actor_id)
.await
@ -90,7 +87,7 @@ impl NodeCache {
#[tracing::instrument(name = "Save node info", fields(actor_id = actor_id.to_string().as_str(), software, version, reg))]
pub(crate) async fn set_info(
&self,
actor_id: Url,
actor_id: IriString,
software: String,
version: String,
reg: bool,
@ -121,7 +118,7 @@ impl NodeCache {
)]
pub(crate) async fn set_instance(
&self,
actor_id: Url,
actor_id: IriString,
title: String,
description: String,
version: String,
@ -155,11 +152,11 @@ impl NodeCache {
)]
pub(crate) async fn set_contact(
&self,
actor_id: Url,
actor_id: IriString,
username: String,
display_name: String,
url: Url,
avatar: Url,
url: IriString,
avatar: IriString,
) -> Result<(), Error> {
self.db
.save_contact(
@ -177,17 +174,18 @@ impl NodeCache {
}
impl Node {
fn new(mut url: Url) -> Self {
url.set_fragment(None);
url.set_query(None);
url.set_path("");
fn new(url: IriString) -> Result<Self, Error> {
let authority = url.authority_str().ok_or(ErrorKind::MissingDomain)?;
let scheme = url.scheme_str();
Node {
base: url,
let base = iri!(format!("{}://{}", scheme, authority));
Ok(Node {
base,
info: None,
instance: None,
contact: None,
}
})
}
fn info(mut self, info: Option<Info>) -> Self {

20
src/data/state.rs
View file

@ -5,7 +5,7 @@ use crate::{
error::Error,
requests::{Breakers, Requests},
};
use activitystreams::url::Url;
use activitystreams::iri_string::types::IriString;
use actix_web::web;
use async_rwlock::RwLock;
use lru::LruCache;
@ -18,7 +18,7 @@ use tracing::info;
pub struct State {
pub(crate) public_key: RsaPublicKey,
private_key: RsaPrivateKey,
object_cache: Arc<RwLock<LruCache<Url, Url>>>,
object_cache: Arc<RwLock<LruCache<IriString, IriString>>>,
node_cache: NodeCache,
breakers: Breakers,
pub(crate) db: Db,
@ -52,22 +52,22 @@ impl State {
name = "Get inboxes for other domains",
fields(
existing_inbox = existing_inbox.to_string().as_str(),
domain
authority
)
)]
pub(crate) async fn inboxes_without(
&self,
existing_inbox: &Url,
domain: &str,
) -> Result<Vec<Url>, Error> {
existing_inbox: &IriString,
authority: &str,
) -> Result<Vec<IriString>, Error> {
Ok(self
.db
.inboxes()
.await?
.iter()
.filter_map(|inbox| {
if let Some(dom) = inbox.domain() {
if inbox != existing_inbox && dom != domain {
if let Some(authority_str) = inbox.authority_str() {
if inbox != existing_inbox && authority_str != authority {
return Some(inbox.clone());
}
}
@ -77,11 +77,11 @@ impl State {
.collect())
}
pub(crate) async fn is_cached(&self, object_id: &Url) -> bool {
pub(crate) async fn is_cached(&self, object_id: &IriString) -> bool {
self.object_cache.read().await.contains(object_id)
}
pub(crate) async fn cache(&self, object_id: Url, actor_id: Url) {
pub(crate) async fn cache(&self, object_id: IriString, actor_id: IriString) {
self.object_cache.write().await.put(object_id, actor_id);
}

144
src/db.rs
View file

@ -1,5 +1,8 @@
use crate::{config::Config, error::Error};
use activitystreams::url::Url;
use crate::{
config::Config,
error::{Error, ErrorKind},
};
use activitystreams::iri_string::types::IriString;
use actix_web::web::Bytes;
use rsa::{
pkcs8::{FromPrivateKey, ToPrivateKey},
@ -41,10 +44,10 @@ impl std::fmt::Debug for Inner {
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub struct Actor {
pub(crate) id: Url,
pub(crate) id: IriString,
pub(crate) public_key: String,
pub(crate) public_key_id: Url,
pub(crate) inbox: Url,
pub(crate) public_key_id: IriString,
pub(crate) inbox: IriString,
pub(crate) saved_at: SystemTime,
}
@ -88,8 +91,8 @@ pub struct Instance {
pub struct Contact {
pub(crate) username: String,
pub(crate) display_name: String,
pub(crate) url: Url,
pub(crate) avatar: Url,
pub(crate) url: IriString,
pub(crate) avatar: IriString,
pub(crate) updated: SystemTime,
}
@ -106,7 +109,10 @@ impl std::fmt::Debug for Contact {
}
impl Inner {
fn connected_by_domain(&self, domains: &[String]) -> impl DoubleEndedIterator<Item = Url> {
fn connected_by_domain(
&self,
domains: &[String],
) -> impl DoubleEndedIterator<Item = IriString> {
let reversed: Vec<_> = domains.iter().map(|s| domain_key(s.as_str())).collect();
self.connected_actor_ids
@ -115,7 +121,7 @@ impl Inner {
.filter_map(|res| res.ok())
.filter_map(url_from_ivec)
.filter_map(move |url| {
let connected_domain = url.domain()?;
let connected_domain = url.authority_str()?;
let connected_rdnn = domain_key(connected_domain);
for rdnn in &reversed {
@ -136,7 +142,7 @@ impl Inner {
.map(|s| String::from_utf8_lossy(&s).to_string())
}
fn connected(&self) -> impl DoubleEndedIterator<Item = Url> {
fn connected(&self) -> impl DoubleEndedIterator<Item = IriString> {
self.connected_actor_ids
.iter()
.values()
@ -156,7 +162,7 @@ impl Inner {
})
}
fn connected_info(&self) -> impl DoubleEndedIterator<Item = (Url, Info)> + '_ {
fn connected_info(&self) -> impl DoubleEndedIterator<Item = (IriString, Info)> + '_ {
self.connected_actor_ids
.iter()
.values()
@ -170,7 +176,7 @@ impl Inner {
})
}
fn connected_instance(&self) -> impl DoubleEndedIterator<Item = (Url, Instance)> + '_ {
fn connected_instance(&self) -> impl DoubleEndedIterator<Item = (IriString, Instance)> + '_ {
self.connected_actor_ids
.iter()
.values()
@ -184,7 +190,7 @@ impl Inner {
})
}
fn connected_contact(&self) -> impl DoubleEndedIterator<Item = (Url, Contact)> + '_ {
fn connected_contact(&self) -> impl DoubleEndedIterator<Item = (IriString, Contact)> + '_ {
self.connected_actor_ids
.iter()
.values()
@ -198,9 +204,9 @@ impl Inner {
})
}
fn is_allowed(&self, domain: &str) -> bool {
let prefix = domain_prefix(domain);
let reverse_domain = domain_key(domain);
fn is_allowed(&self, authority: &str) -> bool {
let prefix = domain_prefix(authority);
let reverse_domain = domain_key(authority);
if self.restricted_mode {
self.allowed_domains
@ -263,11 +269,11 @@ impl Db {
Ok(t)
}
pub(crate) async fn connected_ids(&self) -> Result<Vec<Url>, Error> {
pub(crate) async fn connected_ids(&self) -> Result<Vec<IriString>, Error> {
self.unblock(|inner| Ok(inner.connected().collect())).await
}
pub(crate) async fn save_info(&self, actor_id: Url, info: Info) -> Result<(), Error> {
pub(crate) async fn save_info(&self, actor_id: IriString, info: Info) -> Result<(), Error> {
self.unblock(move |inner| {
let vec = serde_json::to_vec(&info)?;
@ -280,7 +286,7 @@ impl Db {
.await
}
pub(crate) async fn info(&self, actor_id: Url) -> Result<Option<Info>, Error> {
pub(crate) async fn info(&self, actor_id: IriString) -> Result<Option<Info>, Error> {
self.unblock(move |inner| {
if let Some(ivec) = inner.actor_id_info.get(actor_id.as_str().as_bytes())? {
let info = serde_json::from_slice(&ivec)?;
@ -292,14 +298,14 @@ impl Db {
.await
}
pub(crate) async fn connected_info(&self) -> Result<HashMap<Url, Info>, Error> {
pub(crate) async fn connected_info(&self) -> Result<HashMap<IriString, Info>, Error> {
self.unblock(|inner| Ok(inner.connected_info().collect()))
.await
}
pub(crate) async fn save_instance(
&self,
actor_id: Url,
actor_id: IriString,
instance: Instance,
) -> Result<(), Error> {
self.unblock(move |inner| {
@ -314,7 +320,7 @@ impl Db {
.await
}
pub(crate) async fn instance(&self, actor_id: Url) -> Result<Option<Instance>, Error> {
pub(crate) async fn instance(&self, actor_id: IriString) -> Result<Option<Instance>, Error> {
self.unblock(move |inner| {
if let Some(ivec) = inner.actor_id_instance.get(actor_id.as_str().as_bytes())? {
let instance = serde_json::from_slice(&ivec)?;
@ -326,12 +332,16 @@ impl Db {
.await
}
pub(crate) async fn connected_instance(&self) -> Result<HashMap<Url, Instance>, Error> {
pub(crate) async fn connected_instance(&self) -> Result<HashMap<IriString, Instance>, Error> {
self.unblock(|inner| Ok(inner.connected_instance().collect()))
.await
}
pub(crate) async fn save_contact(&self, actor_id: Url, contact: Contact) -> Result<(), Error> {
pub(crate) async fn save_contact(
&self,
actor_id: IriString,
contact: Contact,
) -> Result<(), Error> {
self.unblock(move |inner| {
let vec = serde_json::to_vec(&contact)?;
@ -344,7 +354,7 @@ impl Db {
.await
}
pub(crate) async fn contact(&self, actor_id: Url) -> Result<Option<Contact>, Error> {
pub(crate) async fn contact(&self, actor_id: IriString) -> Result<Option<Contact>, Error> {
self.unblock(move |inner| {
if let Some(ivec) = inner.actor_id_contact.get(actor_id.as_str().as_bytes())? {
let contact = serde_json::from_slice(&ivec)?;
@ -356,12 +366,12 @@ impl Db {
.await
}
pub(crate) async fn connected_contact(&self) -> Result<HashMap<Url, Contact>, Error> {
pub(crate) async fn connected_contact(&self) -> Result<HashMap<IriString, Contact>, Error> {
self.unblock(|inner| Ok(inner.connected_contact().collect()))
.await
}
pub(crate) async fn save_url(&self, url: Url, id: Uuid) -> Result<(), Error> {
pub(crate) async fn save_url(&self, url: IriString, id: Uuid) -> Result<(), Error> {
self.unblock(move |inner| {
inner
.media_id_media_url
@ -393,7 +403,7 @@ impl Db {
.await
}
pub(crate) async fn media_id(&self, url: Url) -> Result<Option<Uuid>, Error> {
pub(crate) async fn media_id(&self, url: IriString) -> Result<Option<Uuid>, Error> {
self.unblock(move |inner| {
if let Some(ivec) = inner.media_url_media_id.get(url.as_str().as_bytes())? {
Ok(uuid_from_ivec(ivec))
@ -404,7 +414,7 @@ impl Db {
.await
}
pub(crate) async fn media_url(&self, id: Uuid) -> Result<Option<Url>, Error> {
pub(crate) async fn media_url(&self, id: Uuid) -> Result<Option<IriString>, Error> {
self.unblock(move |inner| {
if let Some(ivec) = inner.media_id_media_url.get(id.as_bytes())? {
Ok(url_from_ivec(ivec))
@ -442,24 +452,22 @@ impl Db {
self.unblock(|inner| Ok(inner.blocks().collect())).await
}
pub(crate) async fn inboxes(&self) -> Result<Vec<Url>, Error> {
pub(crate) async fn inboxes(&self) -> Result<Vec<IriString>, Error> {
self.unblock(|inner| Ok(inner.connected_actors().map(|actor| actor.inbox).collect()))
.await
}
pub(crate) async fn is_connected(&self, mut id: Url) -> Result<bool, Error> {
id.set_path("");
id.set_query(None);
id.set_fragment(None);
pub(crate) async fn is_connected(&self, base_id: IriString) -> Result<bool, Error> {
let scheme = base_id.scheme_str();
let authority = base_id.authority_str().ok_or(ErrorKind::MissingDomain)?;
let prefix = format!("{}://{}", scheme, authority);
self.unblock(move |inner| {
let connected = inner
.connected_actor_ids
.scan_prefix(id.as_str().as_bytes())
.scan_prefix(prefix.as_bytes())
.values()
.filter_map(|res| res.ok())
.next()
.is_some();
.any(|res| res.is_ok());
Ok(connected)
})
@ -468,8 +476,8 @@ impl Db {
pub(crate) async fn actor_id_from_public_key_id(
&self,
public_key_id: Url,
) -> Result<Option<Url>, Error> {
public_key_id: IriString,
) -> Result<Option<IriString>, Error> {
self.unblock(move |inner| {
if let Some(ivec) = inner
.public_key_id_actor_id
@ -483,7 +491,7 @@ impl Db {
.await
}
pub(crate) async fn actor(&self, actor_id: Url) -> Result<Option<Actor>, Error> {
pub(crate) async fn actor(&self, actor_id: IriString) -> Result<Option<Actor>, Error> {
self.unblock(move |inner| {
if let Some(ivec) = inner.actor_id_actor.get(actor_id.as_str().as_bytes())? {
let actor = serde_json::from_slice(&ivec)?;
@ -511,7 +519,7 @@ impl Db {
.await
}
pub(crate) async fn remove_connection(&self, actor_id: Url) -> Result<(), Error> {
pub(crate) async fn remove_connection(&self, actor_id: IriString) -> Result<(), Error> {
tracing::debug!("Removing Connection: {}", actor_id);
self.unblock(move |inner| {
inner
@ -523,7 +531,7 @@ impl Db {
.await
}
pub(crate) async fn add_connection(&self, actor_id: Url) -> Result<(), Error> {
pub(crate) async fn add_connection(&self, actor_id: IriString) -> Result<(), Error> {
tracing::debug!("Adding Connection: {}", actor_id);
self.unblock(move |inner| {
inner
@ -543,11 +551,11 @@ impl Db {
.remove(connected.as_str().as_bytes())?;
}
for domain in &domains {
for authority in &domains {
inner
.blocked_domains
.insert(domain_key(domain), domain.as_bytes())?;
inner.allowed_domains.remove(domain_key(domain))?;
.insert(domain_key(authority), authority.as_bytes())?;
inner.allowed_domains.remove(domain_key(authority))?;
}
Ok(())
@ -557,8 +565,8 @@ impl Db {
pub(crate) async fn remove_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
self.unblock(move |inner| {
for domain in &domains {
inner.blocked_domains.remove(domain_key(domain))?;
for authority in &domains {
inner.blocked_domains.remove(domain_key(authority))?;
}
Ok(())
@ -568,10 +576,10 @@ impl Db {
pub(crate) async fn add_allows(&self, domains: Vec<String>) -> Result<(), Error> {
self.unblock(move |inner| {
for domain in &domains {
for authority in &domains {
inner
.allowed_domains
.insert(domain_key(domain), domain.as_bytes())?;
.insert(domain_key(authority), authority.as_bytes())?;
}
Ok(())
@ -589,8 +597,8 @@ impl Db {
}
}
for domain in &domains {
inner.allowed_domains.remove(domain_key(domain))?;
for authority in &domains {
inner.allowed_domains.remove(domain_key(authority))?;
}
Ok(())
@ -598,10 +606,10 @@ impl Db {
.await
}
pub(crate) async fn is_allowed(&self, url: Url) -> Result<bool, Error> {
pub(crate) async fn is_allowed(&self, url: IriString) -> Result<bool, Error> {
self.unblock(move |inner| {
if let Some(domain) = url.domain() {
Ok(inner.is_allowed(domain))
if let Some(authority) = url.authority_str() {
Ok(inner.is_allowed(authority))
} else {
Ok(false)
}
@ -639,12 +647,12 @@ impl Db {
}
}
fn domain_key(domain: &str) -> String {
domain.split('.').rev().collect::<Vec<_>>().join(".") + "."
fn domain_key(authority: &str) -> String {
authority.split('.').rev().collect::<Vec<_>>().join(".") + "."
}
fn domain_prefix(domain: &str) -> String {
domain
fn domain_prefix(authority: &str) -> String {
authority
.split('.')
.rev()
.take(2)
@ -653,8 +661,8 @@ fn domain_prefix(domain: &str) -> String {
+ "."
}
fn url_from_ivec(ivec: sled::IVec) -> Option<Url> {
String::from_utf8_lossy(&ivec).parse::<Url>().ok()
fn url_from_ivec(ivec: sled::IVec) -> Option<IriString> {
String::from_utf8_lossy(&ivec).parse::<IriString>().ok()
}
fn uuid_from_ivec(ivec: sled::IVec) -> Option<Uuid> {
@ -664,14 +672,14 @@ fn uuid_from_ivec(ivec: sled::IVec) -> Option<Uuid> {
#[cfg(test)]
mod tests {
use super::Db;
use activitystreams::url::Url;
use activitystreams::iri_string::types::IriString;
use std::future::Future;
#[test]
fn connect_and_verify() {
run(|db| async move {
let example_actor: Url = "http://example.com/actor".parse().unwrap();
let example_sub_actor: Url = "http://example.com/users/fake".parse().unwrap();
let example_actor: IriString = "http://example.com/actor".parse().unwrap();
let example_sub_actor: IriString = "http://example.com/users/fake".parse().unwrap();
db.add_connection(example_actor.clone()).await.unwrap();
assert!(db.is_connected(example_sub_actor).await.unwrap());
})
@ -680,8 +688,8 @@ mod tests {
#[test]
fn disconnect_and_verify() {
run(|db| async move {
let example_actor: Url = "http://example.com/actor".parse().unwrap();
let example_sub_actor: Url = "http://example.com/users/fake".parse().unwrap();
let example_actor: IriString = "http://example.com/actor".parse().unwrap();
let example_sub_actor: IriString = "http://example.com/users/fake".parse().unwrap();
db.add_connection(example_actor.clone()).await.unwrap();
assert!(db.is_connected(example_sub_actor.clone()).await.unwrap());
@ -693,7 +701,7 @@ mod tests {
#[test]
fn connected_actor_in_connected_list() {
run(|db| async move {
let example_actor: Url = "http://example.com/actor".parse().unwrap();
let example_actor: IriString = "http://example.com/actor".parse().unwrap();
db.add_connection(example_actor.clone()).await.unwrap();
assert!(db.connected_ids().await.unwrap().contains(&example_actor));
@ -703,7 +711,7 @@ mod tests {
#[test]
fn disconnected_actor_not_in_connected_list() {
run(|db| async move {
let example_actor: Url = "http://example.com/actor".parse().unwrap();
let example_actor: IriString = "http://example.com/actor".parse().unwrap();
db.add_connection(example_actor.clone()).await.unwrap();
db.remove_connection(example_actor.clone()).await.unwrap();

13
src/error.rs
View file

@ -1,4 +1,4 @@
use activitystreams::{error::DomainError, url::ParseError};
use activitystreams::checked::CheckError;
use actix_web::{
error::{BlockingError, ResponseError},
http::StatusCode,
@ -56,8 +56,11 @@ pub(crate) enum ErrorKind {
#[error("Couldn't parse key, {0}")]
Pkcs8(#[from] rsa::pkcs8::Error),
#[error("Couldn't parse URI, {0}")]
Uri(#[from] ParseError),
#[error("Couldn't parse IRI, {0}")]
ParseIri(#[from] activitystreams::iri_string::validate::Error),
#[error("Couldn't resolve IRI, {0}")]
ResolveIri(#[from] activitystreams::iri_string::resolve::Error),
#[error("Couldn't perform IO, {0}")]
Io(#[from] io::Error),
@ -102,7 +105,7 @@ pub(crate) enum ErrorKind {
CpuCount(#[from] std::num::TryFromIntError),
#[error("{0}")]
HostMismatch(#[from] DomainError),
HostMismatch(#[from] CheckError),
#[error("Invalid or missing content type")]
ContentType,
@ -137,7 +140,7 @@ pub(crate) enum ErrorKind {
#[error("Input is missing a 'id' field")]
MissingId,
#[error("Url is missing a domain")]
#[error("IriString is missing a domain")]
MissingDomain,
#[error("URI is missing domain field")]

10
src/jobs/apub/announce.rs
View file

@ -7,13 +7,13 @@ use crate::{
DeliverMany, JobState,
},
};
use activitystreams::{activity::Announce as AsAnnounce, url::Url};
use activitystreams::{activity::Announce as AsAnnounce, iri_string::types::IriString};
use background_jobs::ActixJob;
use std::{future::Future, pin::Pin};
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub(crate) struct Announce {
object_id: Url,
object_id: IriString,
actor: Actor,
}
@ -27,7 +27,7 @@ impl std::fmt::Debug for Announce {
}
impl Announce {
pub fn new(object_id: Url, actor: Actor) -> Self {
pub fn new(object_id: IriString, actor: Actor) -> Self {
Announce { object_id, actor }
}
@ -49,8 +49,8 @@ impl Announce {
// Generate a type that says "Look at this object"
fn generate_announce(
config: &Config,
activity_id: &Url,
object_id: &Url,
activity_id: &IriString,
object_id: &IriString,
) -> Result<AsAnnounce, Error> {
let announce = AsAnnounce::new(config.generate_url(UrlKind::Actor), object_id.clone());

10
src/jobs/apub/follow.rs
View file

@ -8,7 +8,7 @@ use crate::{
use activitystreams::{
activity::{Accept as AsAccept, Follow as AsFollow},
prelude::*,
url::Url,
iri_string::types::IriString,
};
use background_jobs::ActixJob;
use std::{future::Future, pin::Pin};
@ -62,7 +62,7 @@ impl Follow {
}
// Generate a type that says "I want to follow you"
fn generate_follow(config: &Config, actor_id: &Url, my_id: &Url) -> Result<AsFollow, Error> {
fn generate_follow(config: &Config, actor_id: &IriString, my_id: &IriString) -> Result<AsFollow, Error> {
let follow = AsFollow::new(my_id.clone(), actor_id.clone());
prepare_activity(
@ -75,9 +75,9 @@ fn generate_follow(config: &Config, actor_id: &Url, my_id: &Url) -> Result<AsFol
// Generate a type that says "I accept your follow request"
fn generate_accept_follow(
config: &Config,
actor_id: &Url,
input_id: &Url,
my_id: &Url,
actor_id: &IriString,
input_id: &IriString,
my_id: &IriString,
) -> Result<AsAccept, Error> {
let mut follow = AsFollow::new(actor_id.clone(), my_id.clone());

5
src/jobs/apub/forward.rs
View file

@ -23,7 +23,7 @@ impl Forward {
async fn perform(self, state: JobState) -> Result<(), Error> {
let object_id = self
.input
.object()
.object_unchecked()
.as_single_id()
.ok_or(ErrorKind::MissingId)?;
@ -31,7 +31,8 @@ impl Forward {
state
.job_server
.queue(DeliverMany::new(inboxes, self.input)?).await?;
.queue(DeliverMany::new(inboxes, self.input)?)
.await?;
Ok(())
}

25
src/jobs/apub/mod.rs
View file

@ -7,9 +7,9 @@ use crate::{
use activitystreams::{
activity::{Follow as AsFollow, Undo as AsUndo},
context,
iri_string::types::IriString,
prelude::*,
security,
url::Url,
};
use std::convert::TryInto;
@ -23,16 +23,23 @@ pub(crate) use self::{
announce::Announce, follow::Follow, forward::Forward, reject::Reject, undo::Undo,
};
async fn get_inboxes(state: &State, actor: &Actor, object_id: &Url) -> Result<Vec<Url>, Error> {
let domain = object_id.host().ok_or(ErrorKind::Domain)?.to_string();
async fn get_inboxes(
state: &State,
actor: &Actor,
object_id: &IriString,
) -> Result<Vec<IriString>, Error> {
let authority = object_id
.authority_str()
.ok_or(ErrorKind::Domain)?
.to_string();
state.inboxes_without(&actor.inbox, &domain).await
state.inboxes_without(&actor.inbox, &authority).await
}
fn prepare_activity<T, U, V, Kind>(
mut t: T,
id: impl TryInto<Url, Error = U>,
to: impl TryInto<Url, Error = V>,
id: impl TryInto<IriString, Error = U>,
to: impl TryInto<IriString, Error = V>,
) -> Result<T, Error>
where
T: ObjectExt<Kind> + BaseExt<Kind>,
@ -45,7 +52,11 @@ where
}
// Generate a type that says "I want to stop following you"
fn generate_undo_follow(config: &Config, actor_id: &Url, my_id: &Url) -> Result<AsUndo, Error> {
fn generate_undo_follow(
config: &Config,
actor_id: &IriString,
my_id: &IriString,
) -> Result<AsUndo, Error> {
let mut follow = AsFollow::new(my_id.clone(), actor_id.clone());
follow.set_id(config.generate_url(UrlKind::Activity));

10
src/jobs/contact.rs
View file

@ -3,14 +3,14 @@ use crate::{
error::{Error, ErrorKind},
jobs::JobState,
};
use activitystreams::{object::Image, prelude::*, url::Url};
use activitystreams::{object::Image, prelude::*, iri_string::types::IriString};
use background_jobs::ActixJob;
use std::{future::Future, pin::Pin};
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub(crate) struct QueryContact {
actor_id: Url,
contact_id: Url,
actor_id: IriString,
contact_id: IriString,
}
impl std::fmt::Debug for QueryContact {
@ -23,7 +23,7 @@ impl std::fmt::Debug for QueryContact {
}
impl QueryContact {
pub(crate) fn new(actor_id: Url, contact_id: Url) -> Self {
pub(crate) fn new(actor_id: IriString, contact_id: IriString) -> Self {
QueryContact {
actor_id,
contact_id,
@ -57,7 +57,7 @@ impl QueryContact {
}
}
fn to_contact(contact: AcceptedActors) -> Option<(String, String, Url, Url)> {
fn to_contact(contact: AcceptedActors) -> Option<(String, String, IriString, IriString)> {
let username = contact.preferred_username()?.to_owned();
let display_name = contact.name()?.as_one()?.as_xsd_string()?.to_owned();

6
src/jobs/deliver.rs
View file

@ -1,11 +1,11 @@
use crate::{error::Error, jobs::JobState};
use activitystreams::url::Url;
use activitystreams::iri_string::types::IriString;
use background_jobs::{ActixJob, Backoff};
use std::{future::Future, pin::Pin};
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub(crate) struct Deliver {
to: Url,
to: IriString,
data: serde_json::Value,
}
@ -19,7 +19,7 @@ impl std::fmt::Debug for Deliver {
}
impl Deliver {
pub(crate) fn new<T>(to: Url, data: T) -> Result<Self, Error>
pub(crate) fn new<T>(to: IriString, data: T) -> Result<Self, Error>
where
T: serde::ser::Serialize,
{

6
src/jobs/deliver_many.rs
View file

@ -2,13 +2,13 @@ use crate::{
error::Error,
jobs::{Deliver, JobState},
};
use activitystreams::url::Url;
use activitystreams::iri_string::types::IriString;
use background_jobs::ActixJob;
use futures_util::future::LocalBoxFuture;
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub(crate) struct DeliverMany {
to: Vec<Url>,
to: Vec<IriString>,
data: serde_json::Value,
}
@ -30,7 +30,7 @@ impl std::fmt::Debug for DeliverMany {
}
impl DeliverMany {
pub(crate) fn new<T>(to: Vec<Url>, data: T) -> Result<Self, Error>
pub(crate) fn new<T>(to: Vec<IriString>, data: T) -> Result<Self, Error>
where
T: serde::ser::Serialize,
{

22
src/jobs/instance.rs
View file

@ -1,15 +1,15 @@
use crate::{
config::UrlKind,
error::Error,
error::{Error, ErrorKind},
jobs::{cache_media::CacheMedia, JobState},
};
use activitystreams::url::Url;
use activitystreams::{iri, iri_string::types::IriString};
use background_jobs::ActixJob;
use std::{future::Future, pin::Pin};
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub(crate) struct QueryInstance {
actor_id: Url,
actor_id: IriString,
}
impl std::fmt::Debug for QueryInstance {
@ -21,7 +21,7 @@ impl std::fmt::Debug for QueryInstance {
}
impl QueryInstance {
pub(crate) fn new(actor_id: Url) -> Self {
pub(crate) fn new(actor_id: IriString) -> Self {
QueryInstance { actor_id }
}
@ -40,10 +40,12 @@ impl QueryInstance {
return Ok(());
}
let mut instance_uri = self.actor_id.clone();
instance_uri.set_fragment(None);
instance_uri.set_query(None);
instance_uri.set_path("api/v1/instance");
let authority = self
.actor_id
.authority_str()
.ok_or(ErrorKind::MissingDomain)?;
let scheme = self.actor_id.scheme_str();
let instance_uri = iri!(format!("{}://{}/api/v1/instance", scheme, authority));
let instance = state
.requests
@ -132,6 +134,6 @@ struct Instance {
struct Contact {
username: String,
display_name: String,
url: Url,
avatar: Url,
url: IriString,
avatar: IriString,
}

24
src/jobs/nodeinfo.rs
View file

@ -1,14 +1,14 @@
use crate::{
error::Error,
error::{Error, ErrorKind},
jobs::{JobState, QueryContact},
};
use activitystreams::url::Url;
use activitystreams::{iri, iri_string::types::IriString};
use background_jobs::ActixJob;
use std::{fmt::Debug, future::Future, pin::Pin};
#[derive(Clone, serde::Deserialize, serde::Serialize)]
pub(crate) struct QueryNodeinfo {
actor_id: Url,
actor_id: IriString,
}
impl Debug for QueryNodeinfo {
@ -20,7 +20,7 @@ impl Debug for QueryNodeinfo {
}
impl QueryNodeinfo {
pub(crate) fn new(actor_id: Url) -> Self {
pub(crate) fn new(actor_id: IriString) -> Self {
QueryNodeinfo { actor_id }
}
@ -34,10 +34,12 @@ impl QueryNodeinfo {
return Ok(());
}
let mut well_known_uri = self.actor_id.clone();
well_known_uri.set_fragment(None);
well_known_uri.set_query(None);
well_known_uri.set_path(".well-known/nodeinfo");
let authority = self
.actor_id
.authority_str()
.ok_or(ErrorKind::MissingDomain)?;
let scheme = self.actor_id.scheme_str();
let well_known_uri = iri!(format!("{}://{}/.well-known/nodeinfo", scheme, authority));
let well_known = state
.requests
@ -100,7 +102,7 @@ struct Nodeinfo {
#[derive(serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct Metadata {
staff_accounts: Option<Vec<Url>>,
staff_accounts: Option<Vec<IriString>>,
}
#[derive(serde::Deserialize)]
@ -202,7 +204,7 @@ impl<'de> serde::de::Deserialize<'de> for SupportedNodeinfo {
#[cfg(test)]
mod tests {
use super::{Nodeinfo, WellKnown};
use activitystreams::url::Url;
use activitystreams::iri_string::types::IriString;
const BANANA_DOG: &'static str = r#"{"links":[{"rel":"http://nodeinfo.diaspora.software/ns/schema/2.0","href":"https://banana.dog/nodeinfo/2.0"},{"rel":"http://nodeinfo.diaspora.software/ns/schema/2.1","href":"https://banana.dog/nodeinfo/2.1"}]}"#;
const ASONIX_DOG: &'static str = r#"{"links":[{"rel":"http://nodeinfo.diaspora.software/ns/schema/2.0","href":"https://asonix.dog/nodeinfo/2.0"}]}"#;
@ -224,7 +226,7 @@ mod tests {
assert_eq!(
accounts[0],
"https://soc.hyena.network/users/HyNET"
.parse::<Url>()
.parse::<IriString>()
.unwrap()
);
}

4
src/main.rs
View file

@ -1,7 +1,7 @@
// need this for ructe
#![allow(clippy::needless_borrow)]
use activitystreams::url::Url;
use activitystreams::iri_string::types::IriString;
use actix_web::{web, App, HttpServer};
use console_subscriber::ConsoleLayer;
use opentelemetry::{sdk::Resource, KeyValue};
@ -34,7 +34,7 @@ use self::{
fn init_subscriber(
software_name: &'static str,
opentelemetry_url: Option<&Url>,
opentelemetry_url: Option<&IriString>,
) -> Result<(), anyhow::Error> {
LogTracer::init()?;

10
src/middleware/verifier.rs
View file

@ -4,7 +4,7 @@ use crate::{
error::{Error, ErrorKind},
requests::Requests,
};
use activitystreams::{base::BaseExt, uri, url::Url};
use activitystreams::{base::BaseExt, iri, iri_string::types::IriString};
use actix_web::web;
use http_signature_normalization_actix::{prelude::*, verify::DeprecatedAlgorithm};
use rsa::{hash::Hash, padding::PaddingScheme, pkcs8::FromPublicKey, PublicKey, RsaPublicKey};
@ -23,7 +23,7 @@ impl MyVerify {
signature: String,
signing_string: String,
) -> Result<bool, Error> {
let public_key_id = uri!(key_id);
let public_key_id = iri!(key_id);
let actor_id = if let Some(mut actor_id) = self
.2
@ -85,8 +85,8 @@ impl MyVerify {
enum PublicKeyResponse {
PublicKey {
#[allow(dead_code)]
id: Url,
owner: Url,
id: IriString,
owner: IriString,
#[allow(dead_code)]
public_key_pem: String,
},
@ -94,7 +94,7 @@ enum PublicKeyResponse {
}
impl PublicKeyResponse {
fn actor_id(&self) -> Option<Url> {
fn actor_id(&self) -> Option<IriString> {
match self {
PublicKeyResponse::PublicKey { owner, .. } => Some(owner.clone()),
PublicKeyResponse::Actor(actor) => actor.id_unchecked().cloned(),

50
src/requests.rs
View file

@ -1,8 +1,7 @@
use crate::error::{Error, ErrorKind};
use activitystreams::url::Url;
use activitystreams::iri_string::types::IriString;
use actix_web::{http::header::Date, web::Bytes};
use awc::Client;
use chrono::{DateTime, Utc};
use dashmap::DashMap;
use http_signature_normalization_actix::prelude::*;
use rsa::{hash::Hash, padding::PaddingScheme, RsaPrivateKey};
@ -16,6 +15,7 @@ use std::{
},
time::SystemTime,
};
use time::OffsetDateTime;
use tracing::{debug, info, warn};
use tracing_awc::Tracing;
@ -31,9 +31,9 @@ impl std::fmt::Debug for Breakers {
}
impl Breakers {
fn should_try(&self, url: &Url) -> bool {
if let Some(domain) = url.domain() {
if let Some(breaker) = self.inner.get(domain) {
fn should_try(&self, url: &IriString) -> bool {
if let Some(authority) = url.authority_str() {
if let Some(breaker) = self.inner.get(authority) {
breaker.should_try()
} else {
true
@ -43,10 +43,10 @@ impl Breakers {
}
}
fn fail(&self, url: &Url) {
if let Some(domain) = url.domain() {
fn fail(&self, url: &IriString) {
if let Some(authority) = url.authority_str() {
let should_write = {
if let Some(mut breaker) = self.inner.get_mut(domain) {
if let Some(mut breaker) = self.inner.get_mut(authority) {
breaker.fail();
false
} else {
@ -55,16 +55,16 @@ impl Breakers {
};
if should_write {
let mut breaker = self.inner.entry(domain.to_owned()).or_default();
let mut breaker = self.inner.entry(authority.to_owned()).or_default();
breaker.fail();
}
}
}
fn succeed(&self, url: &Url) {
if let Some(domain) = url.domain() {
fn succeed(&self, url: &IriString) {
if let Some(authority) = url.authority_str() {
let should_write = {
if let Some(mut breaker) = self.inner.get_mut(domain) {
if let Some(mut breaker) = self.inner.get_mut(authority) {
breaker.succeed();
false
} else {
@ -73,7 +73,7 @@ impl Breakers {
};
if should_write {
let mut breaker = self.inner.entry(domain.to_owned()).or_default();
let mut breaker = self.inner.entry(authority.to_owned()).or_default();
breaker.succeed();
}
}
@ -91,8 +91,8 @@ impl Default for Breakers {
#[derive(Debug)]
struct Breaker {
failures: usize,
last_attempt: DateTime<Utc>,
last_success: DateTime<Utc>,
last_attempt: OffsetDateTime,
last_success: OffsetDateTime,
}
impl Breaker {
@ -100,30 +100,30 @@ impl Breaker {
10
}
fn failure_wait() -> chrono::Duration {
chrono::Duration::days(1)
fn failure_wait() -> time::Duration {
time::Duration::days(1)
}
fn should_try(&self) -> bool {
self.failures < Self::failure_threshold()
|| self.last_attempt + Self::failure_wait() < Utc::now()
|| self.last_attempt + Self::failure_wait() < OffsetDateTime::now_utc()
}
fn fail(&mut self) {
self.failures += 1;
self.last_attempt = Utc::now();
self.last_attempt = OffsetDateTime::now_utc();
}
fn succeed(&mut self) {
self.failures = 0;
self.last_attempt = Utc::now();
self.last_success = Utc::now();
self.last_attempt = OffsetDateTime::now_utc();
self.last_success = OffsetDateTime::now_utc();
}
}
impl Default for Breaker {
fn default() -> Self {
let now = Utc::now();
let now = OffsetDateTime::now_utc();
Breaker {
failures: 0,
@ -217,7 +217,7 @@ impl Requests {
where
T: serde::de::DeserializeOwned,
{
let parsed_url = url.parse::<Url>()?;
let parsed_url = url.parse::<IriString>()?;
if !self.breakers.should_try(&parsed_url) {
return Err(ErrorKind::Breaker.into());
@ -274,7 +274,7 @@ impl Requests {
#[tracing::instrument(name = "Fetch Bytes")]
pub(crate) async fn fetch_bytes(&self, url: &str) -> Result<(String, Bytes), Error> {
let parsed_url = url.parse::<Url>()?;
let parsed_url = url.parse::<IriString>()?;
if !self.breakers.should_try(&parsed_url) {
return Err(ErrorKind::Breaker.into());
@ -346,7 +346,7 @@ impl Requests {
"Deliver to Inbox",
fields(self, inbox = inbox.to_string().as_str(), item)
)]
pub(crate) async fn deliver<T>(&self, inbox: Url, item: &T) -> Result<(), Error>
pub(crate) async fn deliver<T>(&self, inbox: IriString, item: &T) -> Result<(), Error>
where
T: serde::ser::Serialize + std::fmt::Debug,
{

46
src/routes/inbox.rs
View file

@ -10,7 +10,8 @@ use crate::{
routes::accepted,
};
use activitystreams::{
activity, base::AnyBase, prelude::*, primitives::OneOrMany, public, url::Url,
activity, base::AnyBase, iri_string::types::IriString, prelude::*, primitives::OneOrMany,
public,
};
use actix_web::{web, HttpResponse};
use http_signature_normalization_actix::prelude::{DigestVerified, SignatureVerified};
@ -79,7 +80,7 @@ pub(crate) async fn route(
fn valid_without_listener(input: &AcceptedActivities) -> Result<bool, Error> {
match input.kind() {
Some(ValidTypes::Follow) => Ok(true),
Some(ValidTypes::Undo) => Ok(single_object(input.object())?.is_kind("Follow")),
Some(ValidTypes::Undo) => Ok(single_object(input.object_unchecked())?.is_kind("Follow")),
_ => Ok(false),
}
}
@ -90,7 +91,7 @@ fn kind_str(base: &AnyBase) -> Result<&str, Error> {
.map_err(Into::into)
}
fn id_string(id: Option<&Url>) -> Result<String, Error> {
fn id_string(id: Option<&IriString>) -> Result<String, Error> {
id.map(|s| s.to_string())
.ok_or(ErrorKind::MissingId)
.map_err(Into::into)
@ -101,11 +102,14 @@ fn single_object(o: &OneOrMany<AnyBase>) -> Result<&AnyBase, Error> {
}
async fn handle_accept(config: &Config, input: AcceptedActivities) -> Result<(), Error> {
let base = single_object(input.object())?.clone();
let base = single_object(input.object_unchecked())?.clone();
let follow = if let Some(follow) = activity::Follow::from_any_base(base)? {
follow
} else {
return Err(ErrorKind::Kind(kind_str(single_object(input.object())?)?.to_owned()).into());
return Err(ErrorKind::Kind(
kind_str(single_object(input.object_unchecked())?)?.to_owned(),
)
.into());
};
if !follow.actor_is(&config.generate_url(UrlKind::Actor)) {
@ -121,11 +125,14 @@ async fn handle_reject(
input: AcceptedActivities,
actor: Actor,
) -> Result<(), Error> {
let base = single_object(input.object())?.clone();
let base = single_object(input.object_unchecked())?.clone();
let follow = if let Some(follow) = activity::Follow::from_any_base(base)? {
follow
} else {
return Err(ErrorKind::Kind(kind_str(single_object(input.object())?)?.to_owned()).into());
return Err(ErrorKind::Kind(
kind_str(single_object(input.object_unchecked())?)?.to_owned(),
)
.into());
};
if !follow.actor_is(&config.generate_url(UrlKind::Actor)) {
@ -144,7 +151,7 @@ async fn handle_undo(
actor: Actor,
is_listener: bool,
) -> Result<(), Error> {
let any_base = single_object(input.object())?.clone();
let any_base = single_object(input.object_unchecked())?.clone();
let undone_object =
AcceptedUndoObjects::from_any_base(any_base)?.ok_or(ErrorKind::ObjectFormat)?;
@ -157,12 +164,13 @@ async fn handle_undo(
}
}
let my_id: Url = config.generate_url(UrlKind::Actor);
let my_id: IriString = config.generate_url(UrlKind::Actor);
if !undone_object.object_is(&my_id) && !undone_object.object_is(&public()) {
return Err(
ErrorKind::WrongActor(id_string(undone_object.object().as_single_id())?).into(),
);
return Err(ErrorKind::WrongActor(id_string(
undone_object.object_unchecked().as_single_id(),
)?)
.into());
}
if !is_listener {
@ -189,13 +197,17 @@ async fn handle_announce(
input: AcceptedActivities,
actor: Actor,
) -> Result<(), Error> {
let object_id = input.object().as_single_id().ok_or(ErrorKind::MissingId)?;
let object_id = input
.object_unchecked()
.as_single_id()
.ok_or(ErrorKind::MissingId)?;
if state.is_cached(object_id).await {
return Err(ErrorKind::Duplicate.into());
}
jobs.queue(Announce::new(object_id.to_owned(), actor)).await?;
jobs.queue(Announce::new(object_id.to_owned(), actor))
.await?;
Ok(())
}
@ -206,10 +218,12 @@ async fn handle_follow(
input: AcceptedActivities,
actor: Actor,
) -> Result<(), Error> {
let my_id: Url = config.generate_url(UrlKind::Actor);
let my_id: IriString = config.generate_url(UrlKind::Actor);
if !input.object_is(&my_id) && !input.object_is(&public()) {
return Err(ErrorKind::WrongActor(id_string(input.object().as_single_id())?).into());
return Err(
ErrorKind::WrongActor(id_string(input.object_unchecked().as_single_id())?).into(),
);
}
jobs.queue(Follow::new(input, actor)).await?;

2
src/routes/nodeinfo.rs
View file

@ -57,7 +57,7 @@ pub(crate) async fn route(
.await
.unwrap_or_default()
.iter()
.filter_map(|listener| listener.domain())
.filter_map(|listener| listener.authority_str())
.map(|s| s.to_owned())
.collect(),
blocks: if config.publish_blocks() {

6
templates/admin.rs.html
View file

@ -1,7 +1,7 @@
@use crate::db::Contact;
@use activitystreams::url::Url;
@use activitystreams::iri_string::types::IriString;
@(contact: &Contact, base: &Url)
@(contact: &Contact, base: &IriString)
<div class="admin">
<div class="left">
@ -12,7 +12,7 @@
<div class="right">
<p class="display-name"><a href="@contact.url">@contact.display_name</a></p>
<p class="username">
@@@contact.username@if let Some(domain) = base.domain() {@@@domain}
@@@contact.username@if let Some(authority) = base.authority_str() {@@@authority}
</p>
</div>
</div>

8
templates/info.rs.html
View file

@ -1,11 +1,11 @@
@use crate::db::Info;
@use activitystreams::url::Url;
@use activitystreams::iri_string::types::IriString;
@(info: &Info, base: &Url)
@(info: &Info, base: &IriString)
<article class="info">
@if let Some(domain) = base.domain() {
<h4 class="padded"><a href="@base">@domain</a></h4>
@if let Some(authority) = base.authority_str() {
<h4 class="padded"><a href="@base">@authority</a></h4>
}
<p class="padded">
Running @info.software, version @info.version.

4
templates/instance.rs.html
View file

@ -1,7 +1,7 @@
@use crate::{db::{Contact, Instance}, templates::admin};
@use activitystreams::url::Url;
@use activitystreams::iri_string::types::IriString;
@(instance: &Instance, software: Option<&str>, contact: Option<&Contact>, base: &Url)
@(instance: &Instance, software: Option<&str>, contact: Option<&Contact>, base: &IriString)
<article class="instance">
<h4 class="padded"><a href="@base">@instance.title</a></h4>