ingest: new utility crate, core: rename deliver::Activity -> Deliverable

apub-core/src/activitypub.rs

@@ -4,7 +4,7 @@ use std::{rc::Rc, sync::Arc};
 
 use url::Url;
 
-static PUBLIC: &'static str = "https://www.w3.org/ns/activitystreams#Public";
+static PUBLIC: &str = "https://www.w3.org/ns/activitystreams#Public";
 
 /// A type that represents a Public Key
 pub trait PublicKey {

apub-core/src/deliver.rs

@@ -44,7 +44,7 @@ pub trait Deliver {
 /// The inverse of [`Deliver`], allows for `item.deliver(&client)` syntax
 ///
 /// ```rust
-/// use apub_core::deliver::Activity;
+/// use apub_core::deliver::Deliverable;
 /// use url::Url;
 ///
 /// #[derive(serde::Deserialize, serde::Serialize)]
@@ -53,10 +53,10 @@ pub trait Deliver {
 ///     // etc...
 /// }
 ///
-/// impl Activity for MyActivity {}
+/// impl Deliverable for MyActivity {}
 /// ```
 #[async_trait::async_trait(?Send)]
-pub trait Activity: serde::ser::Serialize {
+pub trait Deliverable: serde::ser::Serialize {
     /// Deliver the activity to the inbox
     async fn deliver<D: Deliver, S: Session>(
         &self,

apub-core/src/ingest.rs

@@ -156,6 +156,16 @@ pub trait FullRepo {
     }
 }
 
+impl std::fmt::Display for Authority {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Authority::Actor(actor_id) => write!(f, "Actor - '{}'", actor_id),
+            Authority::Server(url) => write!(f, "Server URL - '{}'", url),
+            Authority::None => write!(f, "None"),
+        }
+    }
+}
+
 fn borrow_host(host: &Host<String>) -> Host<&str> {
     match host {
         Host::Ipv4(ip) => Host::Ipv4(*ip),

apub-ingest/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "apub-ingest"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+apub-core = { version = "0.1.0", path = "../apub-core/" }
+async-trait = "0.1.51"
+url = "2"

apub-ingest/src/lib.rs

@@ -0,0 +1,201 @@
+use apub_core::{
+    activitypub::{Activity, DeliverableObject},
+    ingest::{Authority, FullRepo, Ingest},
+    session::Session,
+};
+use url::Url;
+
+pub trait InboxType {
+    fn is_shared(&self) -> bool;
+}
+
+/// Rejects if the Authority does not have permission to act on behalf of Actor
+#[derive(Clone, Debug)]
+pub struct ValidateAuthority<I> {
+    ingest: I,
+}
+
+/// Rejects if a public message was not sent to a shared inbox
+/// Rejects if a private message was sent to a shared inbox
+#[derive(Clone, Debug)]
+pub struct ValidateInbox<I> {
+    ingest: I,
+}
+
+/// Rejects if Activity's host does not match Actor's host
+#[derive(Clone, Debug)]
+pub struct ValidateHosts<I> {
+    ingest: I,
+}
+
+/// Rejects if the Authority does not have permission to act on behalf of Actor
+pub fn validate_authority<I>(ingest: I) -> ValidateAuthority<I> {
+    ValidateAuthority { ingest }
+}
+
+/// Rejects if a public message was not sent to a shared inbox
+/// Rejects if a private message was sent to a shared inbox
+pub fn validate_inbox<I>(ingest: I) -> ValidateInbox<I> {
+    ValidateInbox { ingest }
+}
+
+/// Rejects if Activity's host does not match Actor's host
+pub fn validate_hosts<I>(ingest: I) -> ValidateHosts<I> {
+    ValidateHosts { ingest }
+}
+
+#[derive(Clone, Debug)]
+pub struct AuthorityError {
+    authority: Authority,
+    actor: Url,
+}
+
+#[derive(Clone, Debug)]
+pub struct InboxError;
+
+#[derive(Clone, Debug)]
+pub struct HostError;
+
+#[async_trait::async_trait(?Send)]
+impl<A, I> Ingest<A> for ValidateAuthority<I>
+where
+    A: Activity + 'static,
+    I: Ingest<A>,
+    I::Error: From<AuthorityError>,
+{
+    type Error = I::Error;
+    type ActorId = I::ActorId;
+
+    async fn ingest<R: FullRepo, S: Session>(
+        &self,
+        authority: Authority,
+        actor_id: Self::ActorId,
+        activity: A,
+        repo: R,
+        remote_repo: &R::Remote,
+        session: S,
+    ) -> Result<(), R::Error>
+    where
+        R::Error: From<Self::Error>,
+    {
+        let activity_actor = activity.actor_id();
+
+        let valid = match &authority {
+            Authority::Server(url) => {
+                url.host() == activity_actor.host() && url.port() == activity_actor.port()
+            }
+            Authority::Actor(actor) => actor == activity_actor,
+            Authority::None => false,
+        };
+
+        if !valid {
+            return Err(I::Error::from(AuthorityError {
+                authority,
+                actor: activity_actor.clone(),
+            })
+            .into());
+        }
+
+        self.ingest
+            .ingest(authority, actor_id, activity, repo, remote_repo, session)
+            .await
+    }
+}
+
+#[async_trait::async_trait(?Send)]
+impl<A, I> Ingest<A> for ValidateInbox<I>
+where
+    A: DeliverableObject + 'static,
+    I: Ingest<A>,
+    I::Error: From<InboxError>,
+    I::ActorId: InboxType,
+{
+    type Error = I::Error;
+    type ActorId = I::ActorId;
+
+    async fn ingest<R: FullRepo, S: Session>(
+        &self,
+        authority: Authority,
+        actor_id: Self::ActorId,
+        activity: A,
+        repo: R,
+        remote_repo: &R::Remote,
+        session: S,
+    ) -> Result<(), R::Error>
+    where
+        R::Error: From<Self::Error>,
+    {
+        // delivered a public activity to a user inbox
+        if !actor_id.is_shared() && activity.is_public() {
+            return Err(I::Error::from(InboxError).into());
+        }
+
+        // delivered a private activity to a shared inbox
+        if actor_id.is_shared() && !activity.is_public() {
+            return Err(I::Error::from(InboxError).into());
+        }
+
+        self.ingest
+            .ingest(authority, actor_id, activity, repo, remote_repo, session)
+            .await
+    }
+}
+
+#[async_trait::async_trait(?Send)]
+impl<A, I> Ingest<A> for ValidateHosts<I>
+where
+    A: Activity + 'static,
+    I: Ingest<A>,
+    I::Error: From<HostError>,
+{
+    type Error = I::Error;
+    type ActorId = I::ActorId;
+
+    async fn ingest<R: FullRepo, S: Session>(
+        &self,
+        authority: Authority,
+        actor_id: Self::ActorId,
+        activity: A,
+        repo: R,
+        remote_repo: &R::Remote,
+        session: S,
+    ) -> Result<(), R::Error>
+    where
+        R::Error: From<Self::Error>,
+    {
+        if activity.id().host() != activity.actor_id().host()
+            || activity.id().port() != activity.actor_id().port()
+        {
+            return Err(I::Error::from(HostError).into());
+        }
+
+        self.ingest
+            .ingest(authority, actor_id, activity, repo, remote_repo, session)
+            .await
+    }
+}
+
+impl std::fmt::Display for AuthorityError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "Authority '{}' not permitted to act on behalf of actor '{}'",
+            self.authority, self.actor
+        )
+    }
+}
+impl std::error::Error for AuthorityError {}
+
+impl std::fmt::Display for InboxError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Object delivered to invalid inbox")
+    }
+}
+impl std::error::Error for InboxError {}
+
+impl std::fmt::Display for HostError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Activity and Actor host do not match")
+    }
+}
+impl std::error::Error for HostError {}