xmtp_attachments/
http.rs

1//! Bounded transfer of staged ciphertext.
2
3use std::time::Duration;
4
5use crate::store::AttachmentOptions;
6use crate::{AttachmentError, AttachmentFailureCause as Cause};
7
8#[cfg(not(target_arch = "wasm32"))]
9mod native;
10#[cfg(target_arch = "wasm32")]
11mod wasm;
12
13#[cfg(not(target_arch = "wasm32"))]
14pub use native::Transfer;
15#[cfg(target_arch = "wasm32")]
16pub use wasm::Transfer;
17
18/// Limit for establishing a connection to a storage target or download host.
19#[cfg(not(target_arch = "wasm32"))]
20pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
21/// Limit between bytes during a storage transfer.
22pub const IDLE_TIMEOUT: Duration = Duration::from_secs(60);
23
24/// Values from the backend's signed upload response.
25#[derive(Clone)]
26pub struct UploadRequest {
27    pub method: String,
28    pub url: String,
29    pub headers: Vec<(String, String)>,
30    pub expires_in_seconds: u32,
31}
32
33impl std::fmt::Debug for UploadRequest {
34    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35        let origin = url::Url::parse(&self.url)
36            .map(|url| url.origin().ascii_serialization())
37            .unwrap_or_else(|_| "<invalid>".to_owned());
38        let header_names: Vec<&str> = self.headers.iter().map(|(name, _)| name.as_str()).collect();
39        formatter
40            .debug_struct("UploadRequest")
41            .field("method", &self.method)
42            .field("url", &origin)
43            .field("headers", &header_names)
44            .finish()
45    }
46}
47
48#[derive(Clone, Copy, Debug, PartialEq, Eq)]
49pub enum PutOutcome {
50    Stored,
51    AlreadyStored,
52}
53
54pub(crate) fn put_outcome(status: u16) -> Result<PutOutcome, AttachmentError> {
55    if status == 412 {
56        Ok(PutOutcome::AlreadyStored)
57    } else if (200..300).contains(&status) {
58        Ok(PutOutcome::Stored)
59    } else {
60        Err(AttachmentError::with_http_status(
61            Cause::TargetRejected,
62            status,
63        ))
64    }
65}
66
67/// Compute the bound before a download starts.
68pub fn download_cap(
69    content_length: Option<u64>,
70    snapshot_max_upload_bytes: u64,
71    options: &AttachmentOptions,
72) -> u64 {
73    content_length
74        .unwrap_or(u64::MAX)
75        .min(
76            options
77                .max_download_bytes
78                .unwrap_or(snapshot_max_upload_bytes),
79        )
80        .min(u32::MAX as u64)
81}
82
83pub(crate) fn checked_count(current: u64, added: usize, cap: u64) -> Result<u64, AttachmentError> {
84    let next = current
85        .checked_add(added as u64)
86        .ok_or(AttachmentError::new(Cause::TooLarge))?;
87    if next > cap.min(u32::MAX as u64) {
88        return Err(AttachmentError::new(Cause::TooLarge));
89    }
90    Ok(next)
91}
92
93/// Apply the backend upload URL policy on both targets.
94pub(crate) fn secure_upload_url(url: &url::Url) -> Result<(), AttachmentError> {
95    xmtp_content_types::remote_attachment::content_url_scheme(url.as_str())
96        .map(|_| ())
97        .map_err(|_| AttachmentError::new(Cause::InsecureUrl))
98}
99
100/// Check a published content URL without DNS or transfer policy.
101pub fn check_content_url(raw: &str) -> Result<String, AttachmentError> {
102    xmtp_content_types::remote_attachment::content_url_scheme(raw)
103        .map(str::to_owned)
104        .map_err(|_| AttachmentError::new(Cause::InsecureUrl))
105}
106
107pub(crate) use xmtp_content_types::remote_attachment::is_loopback_name;
108
109#[cfg(test)]
110mod tests {
111    use super::*;
112
113    #[xmtp_common::test(unwrap_try = true)]
114    fn upload_request_debug_hides_signed_values() {
115        let request = UploadRequest {
116            method: "PUT".into(),
117            url: "https://storage.example/path?X-Amz-Signature=query-secret".into(),
118            headers: vec![("authorization".into(), "header-secret".into())],
119            expires_in_seconds: 60,
120        };
121        let debug = format!("{request:?}");
122        assert!(debug.contains("PUT"));
123        assert!(debug.contains("https://storage.example"));
124        assert!(debug.contains("authorization"));
125        assert!(!debug.contains("query-secret"));
126        assert!(!debug.contains("header-secret"));
127        assert!(!debug.contains("/path"));
128    }
129
130    // verifies: ATCH-070
131    #[xmtp_common::test(unwrap_try = true)]
132    fn transfer_deadlines() {
133        #[cfg(not(target_arch = "wasm32"))]
134        assert_eq!(CONNECT_TIMEOUT, Duration::from_secs(30));
135        assert_eq!(IDLE_TIMEOUT, Duration::from_secs(60));
136    }
137
138    // verifies: ATCH-056
139    #[xmtp_common::test(unwrap_try = true)]
140    fn size_cap_least_of() {
141        let mut options = AttachmentOptions::default();
142        assert_eq!(download_cap(Some(10), 100, &options), 10);
143        assert_eq!(download_cap(None, 100, &options), 100);
144        options.max_download_bytes = Some(20);
145        assert_eq!(download_cap(Some(30), 100, &options), 20);
146        options.max_download_bytes = Some(u64::MAX);
147        assert_eq!(download_cap(None, u64::MAX, &options), u32::MAX as u64);
148        assert!(checked_count(9, 2, 10).is_err());
149    }
150
151    #[xmtp_common::test(unwrap_try = true)]
152    fn any_success_status_stores_upload() {
153        for status in [200, 201, 202, 204, 206, 299] {
154            assert_eq!(put_outcome(status)?, PutOutcome::Stored);
155        }
156        assert_eq!(put_outcome(412)?, PutOutcome::AlreadyStored);
157        assert_eq!(put_outcome(500).unwrap_err().cause, Cause::TargetRejected);
158    }
159}
xmtp_attachments/http.rs

xmtp_attachments/
http.rs
