ojak/deliverer/
post.rs

1//! Delivering one activity to one inbox: one attempt of a [`Deliverer`](super::Deliverer) job.
2//!
3//! One signed POST, retried once in the other signature scheme when the
4//! first is refused: a peer that answers 400 or 401 may be one that verifies
5//! only the other. Mastodon 4.7 does the same. The result says which scheme
6//! the peer accepted, so a caller can start there next time, and a failure
7//! says whether trying again later could help.
8
9use crate::client::{Client, RequestError};
10use crate::sig::{Scheme, SenderKey, rfc9421, signature};
11use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
12use std::fmt;
13use std::time::Duration;
14use url::Url;
15
16const ACTIVITY_JSON: &str = "application/activity+json";
17
18/// Why a delivery did not go through.
19#[derive(Debug)]
20pub enum DeliveryError {
21    /// The request was not answered, or was refused before it was sent.
22    Request(RequestError),
23    /// The request could not be signed.
24    Signing(String),
25    /// Something the delivery needs from the application, such as the
26    /// sender's key, could not be read just now.
27    Unavailable(String),
28    /// It was not sent: its destination has failed too often in a row, and
29    /// the circuit breaker holds deliveries to it until its cool-off ends.
30    Held {
31        /// How long until a delivery is let through again.
32        retry_after: Duration,
33    },
34    /// The inbox answered with a status other than success.
35    Status {
36        status: u16,
37        /// What the inbox asked for with `Retry-After`, in seconds.
38        retry_after: Option<Duration>,
39        /// The start of the response body, for the log.
40        body: String,
41    },
42}
43
44impl DeliveryError {
45    /// Whether trying again later cannot help.
46    ///
47    /// A 4xx says the request itself is wrong, except 408 and 429, which say
48    /// it came at a bad time. A refused address or URL stays refused. Anything
49    /// else — 5xx, a timeout, a broken connection — may pass.
50    #[must_use]
51    pub fn is_permanent(&self) -> bool {
52        self.is_permanent_by(permanent_status)
53    }
54
55    /// [`DeliveryError::is_permanent`], with `permanent` saying which
56    /// statuses are.
57    #[must_use]
58    pub fn is_permanent_by(&self, permanent: fn(u16) -> bool) -> bool {
59        match self {
60            Self::Status { status, .. } => permanent(*status),
61            Self::Request(RequestError::Refused(_) | RequestError::InvalidUrl(_))
62            | Self::Signing(_) => true,
63            Self::Request(RequestError::TooLarge | RequestError::Network(_))
64            | Self::Unavailable(_)
65            | Self::Held { .. } => false,
66        }
67    }
68
69    /// The status the inbox answered with, if it answered.
70    #[must_use]
71    pub fn status(&self) -> Option<u16> {
72        match self {
73            Self::Status { status, .. } => Some(*status),
74            _ => None,
75        }
76    }
77
78    /// How long the inbox asked to be left alone, or the circuit breaker
79    /// holds it.
80    #[must_use]
81    pub fn retry_after(&self) -> Option<Duration> {
82        match self {
83            Self::Status { retry_after, .. } => *retry_after,
84            Self::Held { retry_after } => Some(*retry_after),
85            _ => None,
86        }
87    }
88}
89
90/// Whether an inbox answering `status` is answering what will not change: a
91/// 4xx, except 408 and 429, which say the request came at a bad time.
92#[must_use]
93pub fn permanent_status(status: u16) -> bool {
94    (400..500).contains(&status) && !matches!(status, 408 | 429)
95}
96
97impl fmt::Display for DeliveryError {
98    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
99        match self {
100            Self::Request(error) => error.fmt(f),
101            Self::Signing(error) => write!(f, "signing: {error}"),
102            Self::Unavailable(error) => write!(f, "unavailable: {error}"),
103            Self::Held { retry_after } => write!(
104                f,
105                "held: its destination keeps failing; let through in {}s",
106                retry_after.as_secs()
107            ),
108            Self::Status { status, body, .. } => write!(f, "HTTP {status}: {body}"),
109        }
110    }
111}
112
113impl std::error::Error for DeliveryError {}
114
115/// Deliver `body` to `inbox`, signed with `key`, trying `first` and then the
116/// other scheme if `first` is refused. Returns the scheme that was accepted.
117///
118/// # Errors
119///
120/// When neither scheme is accepted, or the request fails for another reason.
121pub async fn deliver(
122    client: &Client,
123    inbox: &Url,
124    body: &[u8],
125    key: &SenderKey,
126    first: Scheme,
127) -> Result<Scheme, DeliveryError> {
128    deliver_with_headers(client, inbox, body, key, first, &[]).await
129}
130
131/// [`deliver`], with `extra` headers sent beside the signature's, such as a
132/// `Collection-Synchronization` header (FEP-8fcf). They are not signed.
133///
134/// # Errors
135///
136/// As [`deliver`].
137pub async fn deliver_with_headers(
138    client: &Client,
139    inbox: &Url,
140    body: &[u8],
141    key: &SenderKey,
142    first: Scheme,
143    extra: &[(String, String)],
144) -> Result<Scheme, DeliveryError> {
145    match attempt(client, inbox, body, key, first, extra).await {
146        Ok(()) => Ok(first),
147        Err(DeliveryError::Status {
148            status: 400 | 401, ..
149        }) => {
150            let second = first.other();
151            attempt(client, inbox, body, key, second, extra).await?;
152            Ok(second)
153        }
154        Err(error) => Err(error),
155    }
156}
157
158async fn attempt(
159    client: &Client,
160    inbox: &Url,
161    body: &[u8],
162    key: &SenderKey,
163    scheme: Scheme,
164    extra: &[(String, String)],
165) -> Result<(), DeliveryError> {
166    let mut headers = HeaderMap::new();
167    insert(&mut headers, "content-type", ACTIVITY_JSON)?;
168    insert(&mut headers, "accept", ACTIVITY_JSON)?;
169    for (name, value) in extra {
170        let header = HeaderName::try_from(name.as_str())
171            .map_err(|error| DeliveryError::Signing(format!("{name}: {error}")))?;
172        let value = HeaderValue::from_str(value)
173            .map_err(|error| DeliveryError::Signing(format!("{name}: {error}")))?;
174        headers.insert(header, value);
175    }
176    match scheme {
177        Scheme::DraftCavage => {
178            // Mastodon covers Content-Type on deliveries, so it is covered here.
179            let signed = signature::sign_request_with_key(
180                "post",
181                inbox.as_str(),
182                body,
183                &key.key_id,
184                &key.private_key,
185                &[("content-type", ACTIVITY_JSON)],
186                chrono::Utc::now().timestamp(),
187            )
188            .map_err(|error| DeliveryError::Signing(error.to_string()))?;
189            insert(&mut headers, "date", &signed.date)?;
190            insert(&mut headers, "digest", &signed.digest)?;
191            insert(&mut headers, "signature", &signed.signature)?;
192        }
193        Scheme::Rfc9421 => {
194            let signed = rfc9421::sign_request(
195                "post",
196                inbox.as_str(),
197                Some(body),
198                &key.key_id,
199                &rfc9421::SigningKey::Rsa(&key.private_key),
200                chrono::Utc::now().timestamp(),
201            )
202            .map_err(|error| DeliveryError::Signing(error.to_string()))?;
203            insert(&mut headers, "signature-input", &signed.signature_input)?;
204            insert(&mut headers, "signature", &signed.signature)?;
205            if let Some(digest) = &signed.content_digest {
206                // Covered by the signature, so it goes out exactly as signed.
207                insert(&mut headers, "content-digest", digest)?;
208            }
209        }
210    }
211
212    let response = client
213        .post(inbox, headers, body.to_vec())
214        .await
215        .map_err(DeliveryError::Request)?;
216    if (200..300).contains(&response.status) {
217        return Ok(());
218    }
219    let retry_after = response
220        .headers
221        .get("retry-after")
222        .and_then(|value| value.to_str().ok())
223        .and_then(|value| value.trim().parse::<u64>().ok())
224        .map(Duration::from_secs);
225    let body: String = String::from_utf8_lossy(&response.body)
226        .chars()
227        .take(200)
228        .collect();
229    Err(DeliveryError::Status {
230        status: response.status,
231        retry_after,
232        body,
233    })
234}
235
236fn insert(headers: &mut HeaderMap, name: &'static str, value: &str) -> Result<(), DeliveryError> {
237    let value = HeaderValue::from_str(value)
238        .map_err(|error| DeliveryError::Signing(format!("{name}: {error}")))?;
239    headers.insert(HeaderName::from_static(name), value);
240    Ok(())
241}
ojak/deliverer/post.rs

ojak/deliverer/
post.rs
