ojak/federation/
inbox.rs

1//! The inbox: what Ojak does with an activity another server POSTs.
2//!
3//! A listener registered for the activity's type receives it typed, from a
4//! sender Ojak has authenticated, with nothing embedded in it that the
5//! sender could not vouch for; *docs/guide/inbox.md* explains it.
6//! Activities are queued, when the application gives Ojak a queue, and run
7//! by an [`InboxWorker`]; otherwise inside the request.
8
9use super::{ActorRef, BoxFuture, Context, Error, Federation, Inner, RequestInfo, empty, signer};
10use crate::deliverer::PortableInbox;
11use crate::origin::same_origin;
12use crate::portable::ApUri;
13use crate::queue::{Job, QueueError, RetryPolicy, SharedQueue};
14use futures_util::StreamExt as _;
15use futures_util::stream;
16use http::{HeaderValue, StatusCode, header};
17use ojak_vocab::json::{FromJson, ToJson, Typed};
18use ojak_vocab::loss::{self, Loss};
19use serde_json::{Value, json};
20use std::future::Future;
21use std::sync::Arc;
22use std::time::Duration;
23use url::Url;
24
25/// The largest activity an inbox reads.
26pub const MAX_BODY: usize = 1024 * 1024;
27
28/// How long an activity's `id` is remembered, so that it is processed once.
29const SEEN_FOR: Duration = Duration::from_secs(24 * 60 * 60);
30
31/// The queue activities wait in, within the application's backend.
32pub const QUEUE: &str = "inbox";
33
34/// An activity, as a listener receives it.
35#[derive(Clone, Debug)]
36pub struct Received<T> {
37    /// The activity, read into the listener's type.
38    pub activity: T,
39    /// The actor Ojak authenticated as its sender. Everything the activity
40    /// says about anyone else is a claim. A portable actor's is its `ap` URI
41    /// with the DID's colons percent-encoded, which is how a `Url` holds one;
42    /// `crate::portable::ApUri::parse` reads it back.
43    pub sender: Url,
44    /// The actor whose inbox it arrived at; `None` for the shared inbox,
45    /// where the recipients are worked out from the addressing.
46    pub recipient: Option<ActorRef>,
47    /// The activity as it arrived, for forwarding.
48    pub document: Value,
49    /// The activity as it arrived, in its sender's spelling, with what the
50    /// sender could not vouch for reduced to references as in `activity`:
51    /// for an application whose handlers read JSON. An activity taken on its
52    /// Linked Data signature is here as JSON-LD reads it, in ojak's spelling,
53    /// since that and not its sender's spelling is what the signature covers.
54    pub vouched: Value,
55    /// The server's actor that delivered it, when that was not its sender's
56    /// server: a relay, or a server forwarding a reply. The activity is taken
57    /// on its sender's proof or Linked Data signature, or as its sender's
58    /// server serves it; this says who passed it on, which is what Mastodon's
59    /// `requested_through_relay?` asks.
60    pub forwarder: Option<Url>,
61    /// What the activity was taken on.
62    pub vouched_by: VouchedBy,
63    /// What reading it into the listener's type did not keep.
64    pub lost: Vec<Loss>,
65}
66
67/// What an activity was taken on: whose word it is that its actor sent it.
68///
69/// An application deciding whether to pass an activity on asks this: a
70/// receiver of the copy cannot check the HTTP signature it arrived with, only
71/// what the activity carries in itself.
72#[derive(Clone, Copy, Debug, Eq, PartialEq)]
73pub enum VouchedBy {
74    /// The HTTP signature of a server on its actor's origin.
75    Signature,
76    /// Its own FEP-8b32 proof, or for a portable actor, its key.
77    Proof,
78    /// Its actor's Linked Data signature, passed on by another server.
79    LinkedData,
80    /// Nothing it came with: it was fetched from where its id says it lives,
81    /// and `document` is that copy.
82    Origin,
83}
84
85impl VouchedBy {
86    fn as_str(self) -> &'static str {
87        match self {
88            Self::Signature => "signature",
89            Self::Proof => "proof",
90            Self::LinkedData => "linked_data",
91            Self::Origin => "origin",
92        }
93    }
94
95    fn parse(value: &str) -> Option<Self> {
96        Some(match value {
97            "signature" => Self::Signature,
98            "proof" => Self::Proof,
99            "linked_data" => Self::LinkedData,
100            "origin" => Self::Origin,
101            _ => return None,
102        })
103    }
104}
105
106/// An authenticated activity, before it is read into a listener's type.
107pub(super) struct Incoming {
108    /// Normalised, with what the sender could not vouch for reduced to
109    /// references.
110    normalized: Value,
111    document: Value,
112    vouched: Value,
113    sender: Url,
114    forwarder: Option<Url>,
115    vouched_by: VouchedBy,
116    recipient: Option<ActorRef>,
117}
118
119pub(super) type ListenerFn<D> =
120    Arc<dyn Fn(Context<D>, Incoming) -> BoxFuture<'static, Result<(), Error>> + Send + Sync>;
121pub(super) type BlockedFn<D> =
122    Arc<dyn Fn(Context<D>, String) -> BoxFuture<'static, Result<bool, Error>> + Send + Sync>;
123pub(super) type UnverifiedFn<D> =
124    Arc<dyn Fn(Context<D>, Value) -> BoxFuture<'static, ()> + Send + Sync>;
125pub(super) type QueueFn<D> = Arc<dyn Fn(&D) -> Option<SharedQueue> + Send + Sync>;
126pub(super) type ForwardFn<D> =
127    Arc<dyn Fn(Context<D>, Forward) -> BoxFuture<'static, Result<(), Error>> + Send + Sync>;
128
129/// A portable inbox this server is a gateway for, as the application
130/// describes it: who receives what arrives there, and every gateway the
131/// actor lists, this one included.
132#[derive(Clone, Debug)]
133pub struct GatewayInbox {
134    pub recipient: ActorRef,
135    pub gateways: Vec<Url>,
136}
137
138/// An activity to forward, as it was authenticated here: the receiver
139/// establishes it for itself, from its proof or from its origin, since the
140/// signature on the forward is this server's.
141#[derive(Clone, Debug)]
142pub struct Forward {
143    pub activity: Value,
144    pub to: ForwardTo,
145}
146
147/// Where an activity is forwarded.
148#[derive(Clone, Debug)]
149pub enum ForwardTo {
150    /// A portable actor's other gateways, as FEP-ef61 asks of the gateway it
151    /// arrived at, this one left out: sent with `Deliverer::send_portable`.
152    Gateways(PortableInbox),
153    /// The members of collections of this server's that the activity is
154    /// addressed to, as ActivityPub asks (§7.1.2) when it concerns something
155    /// of this server's: a reply to a local post, addressed to its author's
156    /// followers, is sent on to those followers, signed by the collection's
157    /// owner.
158    Collections(Vec<CollectionRef>),
159}
160
161/// A collection of this server's, as the application registered it.
162#[derive(Clone, Debug, Eq, PartialEq)]
163pub struct CollectionRef {
164    pub kind: String,
165    pub identifier: String,
166}
167
168/// How long an activity's forwarding is remembered: once is the rule.
169const FORWARDED_FOR: Duration = Duration::from_secs(7 * 24 * 60 * 60);
170
171/// Wrap a typed listener for the registry.
172pub(super) fn listener<D, T, F, Fut, E>(listen: F) -> ListenerFn<D>
173where
174    D: Clone + Send + Sync + 'static,
175    T: Typed + FromJson + ToJson + Send + 'static,
176    F: Fn(Context<D>, Received<T>) -> Fut + Send + Sync + 'static,
177    Fut: Future<Output = Result<(), E>> + Send + 'static,
178    E: Into<Error>,
179{
180    reading(T::TYPE, listen)
181}
182
183/// A listener reading its activity into `T`, named `name` when it cannot.
184fn reading<D, T, F, Fut, E>(name: &'static str, listen: F) -> ListenerFn<D>
185where
186    D: Clone + Send + Sync + 'static,
187    T: FromJson + ToJson + Send + 'static,
188    F: Fn(Context<D>, Received<T>) -> Fut + Send + Sync + 'static,
189    Fut: Future<Output = Result<(), E>> + Send + 'static,
190    E: Into<Error>,
191{
192    Arc::new(move |context: Context<D>, incoming: Incoming| {
193        let activity = match T::from_json(&incoming.normalized) {
194            Ok(activity) => activity,
195            Err(error) => {
196                // Not an error to retry: the same document will read the
197                // same way next time.
198                let error: Error = format!("not a {name}: {error}").into();
199                context.report(&error);
200                return Box::pin(async { Ok(()) }) as BoxFuture<'static, _>;
201            }
202        };
203        let lost = loss::losses(&incoming.normalized, &activity.to_json());
204        let received = Received {
205            activity,
206            sender: incoming.sender,
207            recipient: incoming.recipient,
208            document: incoming.document,
209            vouched: incoming.vouched,
210            forwarder: incoming.forwarder,
211            vouched_by: incoming.vouched_by,
212            lost,
213        };
214        let future = listen(context, received);
215        Box::pin(async move { future.await.map_err(Into::into) })
216    })
217}
218
219/// Wrap a listener for every type no typed listener takes.
220pub(super) fn any_listener<D, F, Fut, E>(listen: F) -> ListenerFn<D>
221where
222    D: Clone + Send + Sync + 'static,
223    F: Fn(Context<D>, Received<ojak_vocab::generated::AnyObject>) -> Fut + Send + Sync + 'static,
224    Fut: Future<Output = Result<(), E>> + Send + 'static,
225    E: Into<Error>,
226{
227    reading("known activity", listen)
228}
229
230/// The `actor` of an activity: an IRI, or an object's `id`.
231pub(super) fn actor_of(document: &Value) -> Option<&str> {
232    match document.get("actor")? {
233        Value::String(actor) => Some(actor),
234        Value::Object(actor) => actor.get("id")?.as_str(),
235        Value::Array(actors) => actors.first().and_then(|actor| match actor {
236            Value::String(actor) => Some(actor.as_str()),
237            actor => actor.get("id")?.as_str(),
238        }),
239        _ => None,
240    }
241}
242
243/// The IRIs a property names: strings, and objects' `id`s.
244fn ids(value: Option<&Value>) -> Vec<&str> {
245    match value {
246        Some(Value::String(id)) => vec![id.as_str()],
247        Some(Value::Object(object)) => object
248            .get("id")
249            .and_then(Value::as_str)
250            .into_iter()
251            .collect(),
252        Some(Value::Array(items)) => items.iter().flat_map(|item| ids(Some(item))).collect(),
253        _ => Vec::new(),
254    }
255}
256
257/// Reduce what the activity embeds to what `sender` can vouch for: an object
258/// whose `id`, or any author it names, is on another origin is replaced by
259/// its `id`, for the listener to fetch from its owner; one with no `id` that
260/// names such an author is dropped.
261pub(super) fn reduce(activity: &mut Value, sender: &Url) {
262    if let Some(object) = activity.get_mut("object") {
263        reduce_value(object, sender);
264    }
265}
266
267fn reduce_value(value: &mut Value, sender: &Url) {
268    match value {
269        Value::Array(items) => {
270            for item in items.iter_mut() {
271                reduce_value(item, sender);
272            }
273            items.retain(|item| !item.is_null());
274        }
275        Value::Object(object) => {
276            let authors_vouched = ["attributedTo", "actor"].iter().all(|key| {
277                ids(object.get(*key))
278                    .into_iter()
279                    .all(|author| same_origin(author, sender.as_str()))
280            });
281            match object.get("id").and_then(Value::as_str) {
282                Some(id) if !(authors_vouched && same_origin(id, sender.as_str())) => {
283                    *value = Value::String(id.to_owned());
284                }
285                None if !authors_vouched => *value = Value::Null,
286                _ => {}
287            }
288        }
289        _ => {}
290    }
291}
292
293/// The listener for a normalised activity's `type`.
294fn listener_for<'a, D>(inner: &'a Inner<D>, normalized: &Value) -> Option<&'a ListenerFn<D>> {
295    let types: Vec<&str> = match normalized.get("type")? {
296        Value::String(kind) => vec![kind.as_str()],
297        Value::Array(kinds) => kinds.iter().filter_map(Value::as_str).collect(),
298        _ => return None,
299    };
300    types
301        .into_iter()
302        .find_map(|kind| inner.listeners.get(kind))
303        .or(inner.fallback_listener.as_ref())
304}
305
306/// Normalise an activity and reduce it to what its sender vouches for — or,
307/// `as_written`, only reduce it.
308fn prepare(document: &Value, sender: &Url, as_written: bool) -> Result<Value, String> {
309    let mut normalized = if as_written {
310        document.clone()
311    } else {
312        ojak_jsonld::normalize_with_cache(
313            &crate::fetch::REGISTRY,
314            document,
315            ojak_jsonld::Limits::default(),
316            &*crate::fetch::CONTEXTS,
317        )
318        .map_err(|error| error.to_string())?
319        .into_document()
320    };
321    if let Some(members) = normalized.as_object_mut() {
322        members.remove("@context");
323    }
324    reduce(&mut normalized, sender);
325    Ok(normalized)
326}
327
328/// What a listener is given of an activity: normalised (or, `as_written`,
329/// as it is) and in its sender's spelling, each reduced to what `sender`
330/// vouches for. An activity taken on its Linked Data signature is given as
331/// `read`, the reading [`read_linked_data`] made of it, both times.
332fn readings(
333    document: &Value,
334    read: Option<&Value>,
335    sender: &Url,
336    as_written: bool,
337) -> Result<(Value, Value), String> {
338    let normalized = match read {
339        Some(read) => prepare(read, sender, true)?,
340        None => prepare(document, sender, as_written)?,
341    };
342    let mut vouched = read.unwrap_or(document).clone();
343    reduce(&mut vouched, sender);
344    Ok((normalized, vouched))
345}
346
347/// An activity forwarded by a server other than its actor's, taken on its
348/// actor's Linked Data signature: how it reads, and the actor that signed.
349///
350/// The reading has to say what the copy as written said of who sent it and
351/// which activity it is, since those are what the inbox went on before it
352/// read it: a copy whose keys say one actor and whose statements another is
353/// refused.
354async fn linked_data<D: Clone + Send + Sync + 'static>(
355    context: &Context<D>,
356    document: &Value,
357    actor: &str,
358) -> Result<(Value, Url), String> {
359    let (sender, registry) = signer::verify_linked_data(context, document, actor).await?;
360    let reading = read_linked_data(document, registry.as_ref())?;
361    if actor_of(&reading) != Some(actor) {
362        return Err(format!("as JSON-LD reads it, it is not by {actor}"));
363    }
364    if reading.get("id") != document.get("id") {
365        return Err("as JSON-LD reads it, it is another activity".into());
366    }
367    Ok((reading, sender))
368}
369
370/// How an activity reads to a JSON-LD processor, in ojak's spelling.
371///
372/// A Linked Data signature covers the statements a document makes, not the
373/// keys it makes them with, so a document can be reworded to read
374/// differently to a reader of keys — `actor` mapped to nothing in its
375/// `@context`, the real actor under `as:actor` — and still verify. Mastodon
376/// compacts such an activity before reading it; ojak normalises it.
377///
378/// `registry` holds the contexts loaded to check the signature, when any
379/// were: the activity is read over what it was signed over. Processed
380/// contexts are cached for ojak's own registry only, which is what the cache
381/// holds.
382fn read_linked_data(
383    document: &Value,
384    registry: Option<&ojak_jsonld::Registry>,
385) -> Result<Value, String> {
386    match registry {
387        Some(registry) => {
388            ojak_jsonld::normalize_with(registry, document, ojak_jsonld::Limits::default())
389        }
390        None => ojak_jsonld::normalize_with_cache(
391            &crate::fetch::REGISTRY,
392            document,
393            ojak_jsonld::Limits::default(),
394            &*crate::fetch::CONTEXTS,
395        ),
396    }
397    .map(ojak_jsonld::Processed::into_document)
398    .map_err(|error| error.to_string())
399}
400
401fn status(code: StatusCode, why: &str) -> http::Response<Vec<u8>> {
402    let mut response = empty(code);
403    if !why.is_empty() {
404        *response.body_mut() = why.as_bytes().to_vec();
405        response.headers_mut().insert(
406            header::CONTENT_TYPE,
407            HeaderValue::from_static("text/plain; charset=utf-8"),
408        );
409    }
410    response
411}
412
413fn accepted() -> http::Response<Vec<u8>> {
414    empty(StatusCode::ACCEPTED)
415}
416
417/// Receive `body`, POSTed to `recipient`'s inbox or the shared one.
418pub(super) async fn receive<D: Clone + Send + Sync + 'static>(
419    context: Context<D>,
420    recipient: Option<ActorRef>,
421    body: &[u8],
422) -> http::Response<Vec<u8>> {
423    receive_at(context, recipient, body, None).await
424}
425
426/// Receive `body`, POSTed to a portable inbox this server is a gateway for:
427/// as any inbox receives it, and forwarded to the actor's other gateways.
428pub(super) async fn receive_at_gateway<D: Clone + Send + Sync + 'static>(
429    context: Context<D>,
430    inbox: ApUri,
431    gateway: GatewayInbox,
432    body: &[u8],
433) -> http::Response<Vec<u8>> {
434    let recipient = Some(gateway.recipient.clone());
435    receive_at(context, recipient, body, Some((inbox, gateway.gateways))).await
436}
437
438async fn receive_at<D: Clone + Send + Sync + 'static>(
439    context: Context<D>,
440    recipient: Option<ActorRef>,
441    body: &[u8],
442    gateway: Option<(ApUri, Vec<Url>)>,
443) -> http::Response<Vec<u8>> {
444    let inner = context.inner.federation.clone();
445    if body.len() > MAX_BODY {
446        return empty(StatusCode::PAYLOAD_TOO_LARGE);
447    }
448    let mut document: Value = match serde_json::from_slice(body) {
449        Ok(document @ Value::Object(_)) => document,
450        _ => return status(StatusCode::BAD_REQUEST, "not a JSON object"),
451    };
452    let Some(actor) = actor_of(&document).map(str::to_owned) else {
453        return status(StatusCode::BAD_REQUEST, "no actor");
454    };
455
456    let portable = ApUri::parse(&actor);
457
458    // A request signed with a key on a blocked server is refused, 403, before
459    // the key is looked for, as Mastodon's `keypair_from_key_id` refuses it,
460    // and with the body its `require_actor_signature!` renders.
461    if portable.is_none() {
462        match signer::blocked_key_id(&context, body).await {
463            Ok(Some(key_id)) => {
464                let mut response = empty(StatusCode::FORBIDDEN);
465                *response.body_mut() =
466                    json!({ "error": format!("Public key not found for key {key_id}") })
467                        .to_string()
468                        .into_bytes();
469                response.headers_mut().insert(
470                    header::CONTENT_TYPE,
471                    HeaderValue::from_static("application/json; charset=utf-8"),
472                );
473                return response;
474            }
475            Ok(None) => {}
476            Err(error) => {
477                context.report(&error);
478                return empty(StatusCode::INTERNAL_SERVER_ERROR);
479            }
480        }
481    }
482
483    // A blocked server costs nothing: no key is fetched for it. A portable
484    // actor has no server, and is blocked by its DID.
485    if let Some(blocked) = &inner.blocked {
486        let host = match &portable {
487            Some(uri) => uri.did().to_owned(),
488            None => Url::parse(&actor)
489                .ok()
490                .and_then(|url| url.host_str().map(str::to_owned))
491                .unwrap_or_default(),
492        };
493        match blocked(context.clone(), host).await {
494            Ok(true) => return accepted(),
495            Ok(false) => {}
496            Err(error) => {
497                context.report(&error);
498                return empty(StatusCode::INTERNAL_SERVER_ERROR);
499            }
500        }
501    }
502
503    // How an activity taken on its Linked Data signature reads (see
504    // `read_linked_data`); `None` for every other activity.
505    let mut read: Option<Value> = None;
506    // Who delivered it, when that was not the actor's server.
507    let mut forwarder: Option<Url> = None;
508    let mut vouched_by = VouchedBy::Signature;
509    let sender = if let Some(uri) = &portable {
510        // A portable actor is vouched for by its key alone: the HTTP
511        // signature, if there is one, is the gateway's that sent it.
512        let resolver = inner
513            .signed_fetch
514            .as_ref()
515            .map(|settings| settings.fetcher(context.data()));
516        let resolver = resolver
517            .as_deref()
518            .and_then(crate::fetch::Fetcher::resolver);
519        vouched_by = VouchedBy::Proof;
520        match crate::portable::verify_by(&document, uri.did(), resolver).await {
521            Ok(()) => match Url::parse(&uri.encoded()) {
522                Ok(sender) => sender,
523                Err(error) => return status(StatusCode::BAD_REQUEST, &error.to_string()),
524            },
525            Err(unproven) => {
526                if let Some(hook) = &inner.on_unverified {
527                    hook(context.clone(), document.clone()).await;
528                }
529                // Unlike a server's actor, a DID does not go away and take
530                // its key with it, so a Delete is refused like anything else.
531                return status(StatusCode::UNAUTHORIZED, &format!("proof: {unproven}"));
532            }
533        }
534    } else {
535        let signed = signer::authenticate(&context, body).await;
536        forwarder = signed
537            .as_ref()
538            .ok()
539            .filter(|signer| !same_origin(&actor, signer.as_str()))
540            .cloned();
541        match signed {
542            Ok(sender) if same_origin(&actor, sender.as_str()) => sender,
543            // A forwarded activity's Linked Data signature, when it is to be
544            // checked before its proof (`linked_data_before_proof`).
545            signed => match if signed.is_ok() && inner.linked_data_first {
546                Some(linked_data(&context, &document, &actor).await)
547            } else {
548                None
549            } {
550                Some(Ok((reading, sender))) => {
551                    read = Some(reading);
552                    vouched_by = VouchedBy::LinkedData;
553                    sender
554                }
555                checked => match signer::prove(&context, &document, &actor).await {
556                    Ok(sender) => {
557                        vouched_by = VouchedBy::Proof;
558                        sender
559                    }
560                    // Signed by a server other than the actor's: forwarded, by
561                    // a relay, a gateway or a server passing on a reply. The
562                    // forwarder vouches for nothing of the actor's. The actor's
563                    // own Linked Data signature, as Mastodon signs what relays
564                    // pass on, vouches for it; failing that, the activity is
565                    // taken from where its id says it lives instead. Only a
566                    // signed request gets this far, so an unsigned POST cannot
567                    // make this server fetch.
568                    Err(unproven) if signed.is_ok() => {
569                        let linked = match checked {
570                            Some(linked) => linked,
571                            None => linked_data(&context, &document, &actor).await,
572                        };
573                        match linked {
574                            Ok((reading, sender)) => {
575                                read = Some(reading);
576                                vouched_by = VouchedBy::LinkedData;
577                                sender
578                            }
579                            Err(unsigned) => {
580                                match signer::establish(&context, &document, &actor).await {
581                                    Ok((established, sender)) => {
582                                        document = established;
583                                        vouched_by = VouchedBy::Origin;
584                                        sender
585                                    }
586                                    // Dropped, and answered 202 as Mastodon answers
587                                    // it: the forwarder's own signature verified, so
588                                    // there is nothing for it to retry, and a 401 had
589                                    // forwarders sending the same activity dozens of
590                                    // times. Mastodon accepts a delivery once its
591                                    // signature verifies and drops a relayed activity
592                                    // it cannot verify later
593                                    // (ActivityPub::ProcessActivityService). The reason
594                                    // goes back in the body, for the application to log.
595                                    Err(unfetched) => {
596                                        return status(
597                                            StatusCode::ACCEPTED,
598                                            &format!(
599                                                "dropped: forwarded; proof: {unproven}; \
600                                         linked data: {unsigned}; origin: {unfetched}"
601                                            ),
602                                        );
603                                    }
604                                }
605                            }
606                        }
607                    }
608                    Err(unproven) => {
609                        let unsigned = signed.err().unwrap_or_default();
610                        if let Some(hook) = &inner.on_unverified {
611                            hook(context.clone(), document.clone()).await;
612                        }
613                        // A Delete that does not verify is most often one whose
614                        // actor is gone, key and all; the server would retry it
615                        // until told otherwise.
616                        if document.get("type").and_then(Value::as_str) == Some("Delete") {
617                            return accepted();
618                        }
619                        return status(
620                            StatusCode::UNAUTHORIZED,
621                            &format!("signature: {unsigned}; proof: {unproven}"),
622                        );
623                    }
624                },
625            },
626        }
627    };
628
629    // A server vouches for its own actors and activities, and nobody else's.
630    if !same_origin(&actor, sender.as_str()) {
631        return status(
632            StatusCode::UNAUTHORIZED,
633            &format!("actor {actor} is not on the origin of {sender}"),
634        );
635    }
636    let id = document
637        .get("id")
638        .and_then(Value::as_str)
639        .map(str::to_owned);
640    if let Some(id) = &id
641        && !same_origin(id, sender.as_str())
642    {
643        return status(
644            StatusCode::UNAUTHORIZED,
645            &format!("activity {id} is not on the origin of {sender}"),
646        );
647    }
648
649    let (normalized, vouched) =
650        match readings(&document, read.as_ref(), &sender, inner.read_as_written) {
651            Ok(readings) => readings,
652            Err(error) => return status(StatusCode::BAD_REQUEST, &error),
653        };
654
655    // Forwarded before anything else is decided, whether or not a listener
656    // wants it: the other gateways keep the actor's data too.
657    let origin = context.origin().to_string();
658    if let Some(id) = &id {
659        if let Some((inbox, gateways)) = gateway {
660            let ours = context.origin().clone();
661            let others: Vec<Url> = gateways
662                .into_iter()
663                .filter(|gateway| !same_origin(gateway.as_str(), ours.as_str()))
664                .collect();
665            if !others.is_empty() {
666                let to = ForwardTo::Gateways(PortableInbox {
667                    inbox,
668                    gateways: others,
669                });
670                forward(&context, &origin, id, &document, to).await;
671            }
672        }
673        // Who it is addressed to is read as everything else is; what is
674        // forwarded is the activity as it came, its signature intact.
675        let collections = forwarded_to(&context, read.as_ref().unwrap_or(&document));
676        if !collections.is_empty() {
677            let to = ForwardTo::Collections(collections);
678            forward(&context, &origin, id, &document, to).await;
679        }
680    }
681
682    if listener_for(&inner, &normalized).is_none() {
683        return accepted();
684    }
685
686    // Once each: the id is remembered under the origin it arrived at, so
687    // that the same activity delivered to two instances in one process is
688    // processed by both, and with a digest of what arrived, so that another
689    // activity under an id already seen is still processed. Mastodon itself
690    // remembers no ids, and names some activities alike: every Reject of a
691    // QuoteRequest an account sends has the one id.
692    let digest = id.as_ref().map(|_| {
693        use base64::Engine as _;
694        use sha2::Digest as _;
695        let bytes = serde_json::to_vec(&document).unwrap_or_default();
696        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(sha2::Sha256::digest(&bytes))
697    });
698    let seen = id
699        .as_deref()
700        .zip(digest.as_deref())
701        .map(|(id, digest)| ["ojak", "inbox", origin.as_str(), id, digest]);
702    if let (Some(seen), Some(settings)) = (&seen, &inner.signed_fetch) {
703        match settings
704            .kv(context.data())
705            .insert(seen, Value::Bool(true), Some(SEEN_FOR))
706            .await
707        {
708            Ok(true) => {}
709            Ok(false) => return accepted(),
710            Err(error) => context.report(&Error::from(error)),
711        }
712    }
713
714    let queue = inner
715        .inbox_queue
716        .as_ref()
717        .and_then(|queue| queue(context.data()));
718    let outcome = match queue {
719        Some(queue) => {
720            let mut payload = json!({
721                "document": document,
722                "sender": sender.as_str(),
723                "recipient": recipient.as_ref().map(|r| json!({"kind": r.kind, "identifier": r.identifier})),
724                "origin": origin,
725                "vouched_by": vouched_by.as_str(),
726            });
727            if let Some(read) = read {
728                payload["read"] = read;
729            }
730            if let Some(forwarder) = &forwarder {
731                payload["forwarder"] = Value::from(forwarder.as_str());
732            }
733            queue
734                .enqueue(QUEUE, vec![payload])
735                .await
736                .map_err(Error::from)
737        }
738        None => {
739            let listener = listener_for(&inner, &normalized)
740                .expect("found above")
741                .clone();
742            listener(
743                context.clone(),
744                Incoming {
745                    normalized,
746                    document,
747                    vouched,
748                    sender,
749                    forwarder,
750                    vouched_by,
751                    recipient,
752                },
753            )
754            .await
755        }
756    };
757    match outcome {
758        Ok(()) => accepted(),
759        Err(error) => {
760            context.report(&error);
761            // Not processed, so not seen: the sender's retry is processed.
762            if let (Some(seen), Some(settings)) = (&seen, &inner.signed_fetch) {
763                let _ = settings.kv(context.data()).delete(seen).await;
764            }
765            empty(StatusCode::INTERNAL_SERVER_ERROR)
766        }
767    }
768}
769
770/// Hand `activity` to the application to forward `to` where it says: once
771/// for each kind of forward, however many times the activity arrives, so
772/// that servers forwarding to each other stop.
773async fn forward<D: Clone + Send + Sync + 'static>(
774    context: &Context<D>,
775    origin: &str,
776    id: &str,
777    activity: &Value,
778    to: ForwardTo,
779) {
780    let inner = &context.inner.federation;
781    let (Some(hook), Some(settings)) = (&inner.forward, &inner.signed_fetch) else {
782        return;
783    };
784    let kind = match to {
785        ForwardTo::Gateways(_) => "gateways",
786        ForwardTo::Collections(_) => "collections",
787    };
788    let key = ["ojak", "forwarded", kind, origin, id];
789    match settings
790        .kv(context.data())
791        .insert(&key, Value::Bool(true), Some(FORWARDED_FOR))
792        .await
793    {
794        Ok(true) => {}
795        Ok(false) => return,
796        Err(error) => {
797            // Not knowing whether it was forwarded, it is not forwarded
798            // again: once is the rule, and the others forward too.
799            context.report(&Error::from(error));
800            return;
801        }
802    }
803    let forward = Forward {
804        activity: activity.clone(),
805        to,
806    };
807    if let Err(error) = hook(context.clone(), forward).await {
808        context.report(&error);
809    }
810}
811
812/// The collections of ours `activity` is to be forwarded to (ActivityPub
813/// §7.1.2): those it is addressed to, when it concerns something of ours,
814/// in its `object`, `target`, `inReplyTo` or `tag`, or in those of an object
815/// it embeds.
816fn forwarded_to<D: Clone + Send + Sync + 'static>(
817    context: &Context<D>,
818    activity: &Value,
819) -> Vec<CollectionRef> {
820    if context.inner.federation.forward.is_none() {
821        return Vec::new();
822    }
823    let mut collections = Vec::new();
824    for key in ["to", "cc", "audience"] {
825        for iri in ids(activity.get(key)) {
826            if let Some(super::Route::Collection { kind, identifier }) = context.parse_uri(iri) {
827                let collection = CollectionRef { kind, identifier };
828                if !collections.contains(&collection) {
829                    collections.push(collection);
830                }
831            }
832        }
833    }
834    if collections.is_empty() {
835        return collections;
836    }
837    let concerning = |value: &Value| {
838        ["object", "target", "inReplyTo", "tag"]
839            .iter()
840            .flat_map(|key| ids(value.get(*key)))
841            .any(|iri| context.parse_uri(iri).is_some())
842    };
843    let embedded = match activity.get("object") {
844        Some(Value::Array(items)) => items
845            .iter()
846            .any(|item| item.is_object() && concerning(item)),
847        Some(item @ Value::Object(_)) => concerning(item),
848        _ => false,
849    };
850    if concerning(activity) || embedded {
851        collections
852    } else {
853        Vec::new()
854    }
855}
856
857/// How an [`InboxWorker`] runs.
858#[derive(Clone, Debug)]
859pub struct InboxWorkerConfig {
860    /// How many activities one claim takes.
861    pub batch: usize,
862    /// How long a claimed activity is held before another worker may take it.
863    pub lease: Duration,
864    /// Listeners running at once.
865    pub concurrency: usize,
866    pub retry: RetryPolicy,
867    /// The longest the worker sleeps with nothing due.
868    pub idle_poll: Duration,
869}
870
871impl Default for InboxWorkerConfig {
872    fn default() -> Self {
873        Self {
874            batch: 32,
875            lease: Duration::from_secs(300),
876            concurrency: 4,
877            retry: RetryPolicy::default(),
878            idle_poll: Duration::from_secs(5),
879        }
880    }
881}
882
883/// Runs the listeners for activities queued in an inbox's queue.
884pub struct InboxWorker<D> {
885    federation: Federation<D>,
886    data: D,
887    queue: SharedQueue,
888    config: InboxWorkerConfig,
889}
890
891impl<D: Clone + Send + Sync + 'static> InboxWorker<D> {
892    /// A worker for the activities queued in `queue`, running listeners with
893    /// `data`.
894    #[must_use]
895    pub fn new(federation: Federation<D>, data: D, queue: SharedQueue) -> Self {
896        Self {
897            federation,
898            data,
899            queue,
900            config: InboxWorkerConfig::default(),
901        }
902    }
903
904    #[must_use]
905    pub fn with_config(mut self, config: InboxWorkerConfig) -> Self {
906        self.config = config;
907        self
908    }
909
910    /// Run queued activities until `stop` completes, finishing the batch in
911    /// hand first.
912    pub async fn run_until(&self, stop: impl Future<Output = ()>) {
913        tokio::pin!(stop);
914        loop {
915            tokio::select! {
916                biased;
917                () = &mut stop => return,
918                () = std::future::ready(()) => {}
919            }
920            let pause = match self.run_once().await {
921                Ok(0) => {
922                    let due = self.queue.next_due(QUEUE).await.ok().flatten();
923                    due.map_or(self.config.idle_poll, |due| due.min(self.config.idle_poll))
924                }
925                Ok(_) => continue,
926                Err(_) => Duration::from_secs(5),
927            };
928            tokio::select! {
929                () = &mut stop => return,
930                () = tokio::time::sleep(pause) => {}
931            }
932        }
933    }
934
935    /// Claim one batch and run it. Returns how many activities it held.
936    ///
937    /// # Errors
938    ///
939    /// When the queue cannot be read.
940    pub async fn run_once(&self) -> Result<usize, QueueError> {
941        let batch = self
942            .queue
943            .claim(QUEUE, self.config.batch, self.config.lease)
944            .await?;
945        let count = batch.len();
946        stream::iter(batch)
947            .for_each_concurrent(self.config.concurrency, |job| self.process(job))
948            .await;
949        Ok(count)
950    }
951
952    async fn process(&self, job: Job) {
953        let outcome = self.run(&job.payload).await;
954        // A queue that cannot be written to leaves the job leased; it comes
955        // back when the lease lapses.
956        let _ = match outcome {
957            Ok(()) => self.queue.complete(&job.id).await,
958            Err(error) => {
959                let message = error.to_string();
960                match self.config.retry.delay(job.attempts + 1) {
961                    Some(delay) => self.queue.retry(&job.id, delay, &message).await,
962                    None => self.queue.fail(&job.id, &message).await,
963                }
964            }
965        };
966    }
967
968    async fn run(&self, payload: &Value) -> Result<(), Error> {
969        let document = payload.get("document").cloned().ok_or("no document")?;
970        let sender = Url::parse(
971            payload
972                .get("sender")
973                .and_then(Value::as_str)
974                .ok_or("no sender")?,
975        )?;
976        let origin = Url::parse(
977            payload
978                .get("origin")
979                .and_then(Value::as_str)
980                .ok_or("no origin")?,
981        )?;
982        let recipient = payload.get("recipient").and_then(|r| {
983            Some(ActorRef::new(
984                r.get("kind")?.as_str()?,
985                r.get("identifier")?.as_str()?,
986            ))
987        });
988        let forwarder = payload
989            .get("forwarder")
990            .and_then(Value::as_str)
991            .and_then(|forwarder| Url::parse(forwarder).ok());
992        // Queued before it was noted: a reading is a Linked Data signature's.
993        let vouched_by = payload
994            .get("vouched_by")
995            .and_then(Value::as_str)
996            .and_then(VouchedBy::parse)
997            .unwrap_or(if payload.get("read").is_some() {
998                VouchedBy::LinkedData
999            } else {
1000                VouchedBy::Signature
1001            });
1002        let (normalized, vouched) = readings(
1003            &document,
1004            payload.get("read"),
1005            &sender,
1006            self.federation.inner.read_as_written,
1007        )?;
1008        let inner = &self.federation.inner;
1009        let Some(listener) = listener_for(inner, &normalized) else {
1010            return Ok(());
1011        };
1012        let context = Context::new(
1013            inner.clone(),
1014            self.data.clone(),
1015            origin,
1016            None::<RequestInfo>,
1017        );
1018        let outcome = listener(
1019            context.clone(),
1020            Incoming {
1021                normalized,
1022                document,
1023                vouched,
1024                sender,
1025                forwarder,
1026                vouched_by,
1027                recipient,
1028            },
1029        )
1030        .await;
1031        if let Err(error) = &outcome {
1032            context.report(error);
1033        }
1034        outcome
1035    }
1036}
1037
1038#[cfg(test)]
1039mod tests {
1040    use super::*;
1041
1042    fn sender() -> Url {
1043        Url::parse("https://a.example/users/alice").unwrap()
1044    }
1045
1046    #[test]
1047    fn what_the_sender_cannot_vouch_for_is_reduced_to_a_reference() {
1048        let mut activity = json!({
1049            "type": "Create",
1050            "actor": "https://a.example/users/alice",
1051            "object": {
1052                "id": "https://b.example/notes/1",
1053                "type": "Note",
1054                "attributedTo": "https://b.example/users/bob",
1055                "content": "words put in bob's mouth"
1056            }
1057        });
1058        reduce(&mut activity, &sender());
1059        assert_eq!(activity["object"], "https://b.example/notes/1");
1060
1061        // On the sender's origin, but naming an author elsewhere.
1062        let mut activity = json!({
1063            "type": "Create",
1064            "object": {
1065                "id": "https://a.example/notes/2",
1066                "type": "Note",
1067                "attributedTo": ["https://a.example/users/alice", "https://b.example/users/bob"]
1068            }
1069        });
1070        reduce(&mut activity, &sender());
1071        assert_eq!(activity["object"], "https://a.example/notes/2");
1072
1073        // Anonymous, and naming someone else as its actor: dropped.
1074        let mut activity = json!({
1075            "type": "Undo",
1076            "object": [{"type": "Follow", "actor": "https://b.example/users/bob"}]
1077        });
1078        reduce(&mut activity, &sender());
1079        assert_eq!(activity["object"], json!([]));
1080    }
1081
1082    #[test]
1083    fn what_the_sender_vouches_for_is_kept() {
1084        let object = json!({
1085            "id": "https://a.example/notes/3",
1086            "type": "Note",
1087            "attributedTo": {"id": "https://a.example/users/alice"},
1088            "content": "mine"
1089        });
1090        let mut activity = json!({"type": "Create", "object": object.clone()});
1091        reduce(&mut activity, &sender());
1092        assert_eq!(activity["object"], object);
1093    }
1094}
ojak/federation/inbox.rs

ojak/federation/
inbox.rs
