void_core/store/
ipfs.rs

1//! IPFS-backed remote object store.
2//!
3//! Implements `RemoteStore` for IPFS via Kubo API or public gateway.
4
5use std::io::Read;
6use std::path::Path;
7use std::time::Duration;
8
9use serde::Deserialize;
10use void_crypto::EncryptedBlob;
11
12use super::RemoteStore;
13use crate::cid::{self, VoidCid};
14use crate::{Result, VoidError};
15
16// ---------------------------------------------------------------------------
17// HTTP helpers
18// ---------------------------------------------------------------------------
19
20/// Extract a human-readable error from a ureq error.
21fn extract_ureq_error(err: ureq::Error) -> String {
22    match err {
23        ureq::Error::Status(status, response) => {
24            let mut body = String::new();
25            if response.into_reader().read_to_string(&mut body).is_ok() {
26                if let Ok(parsed) = serde_json::from_str::<KuboError>(&body) {
27                    return parsed.message;
28                }
29                let trimmed = body.trim();
30                if !trimmed.is_empty() && trimmed.len() < 512 {
31                    return format!("HTTP {status}: {trimmed}");
32                }
33            }
34            format!("HTTP {status}")
35        }
36        ureq::Error::Transport(t) => t.to_string(),
37    }
38}
39
40#[derive(Deserialize)]
41struct KuboError {
42    #[serde(rename = "Message")]
43    message: String,
44}
45
46/// Read an HTTP response body as bytes, tolerating ureq chunked-decoding errors.
47fn read_response_bytes(reader: &mut impl Read) -> Vec<u8> {
48    let mut buf = Vec::new();
49    let mut tmp = [0u8; 8192];
50    loop {
51        match reader.read(&mut tmp) {
52            Ok(0) => break,
53            Ok(n) => buf.extend_from_slice(&tmp[..n]),
54            Err(_) => break, // trailer parse error — data already read
55        }
56    }
57    buf
58}
59
60// ---------------------------------------------------------------------------
61// IpfsBackend + IpfsStore
62// ---------------------------------------------------------------------------
63
64#[derive(Debug, Clone)]
65pub enum IpfsBackend {
66    Kubo { api: String },
67    Gateway { base: String },
68}
69
70#[derive(Debug, Clone)]
71pub struct IpfsStore {
72    backend: IpfsBackend,
73    agent: ureq::Agent,
74}
75
76#[derive(Deserialize)]
77struct BlockPutResponse {
78    #[serde(rename = "Key")]
79    key: String,
80}
81
82impl IpfsStore {
83    pub fn new(backend: IpfsBackend, timeout: Duration) -> Self {
84        let agent = ureq::AgentBuilder::new().timeout(timeout).build();
85        Self { backend, agent }
86    }
87
88    // -----------------------------------------------------------------------
89    // Low-level HTTP ops (private)
90    // -----------------------------------------------------------------------
91
92    fn fetch_raw_http(&self, cid: &str) -> Result<Vec<u8>> {
93        match &self.backend {
94            IpfsBackend::Kubo { api } => {
95                let url = format!("{}/api/v0/block/get?arg={}", api.trim_end_matches('/'), cid);
96                let response = self
97                    .agent
98                    .post(&url)
99                    .call()
100                    .map_err(|e| VoidError::Network(extract_ureq_error(e)))?;
101                let mut reader = response.into_reader();
102                let out = read_response_bytes(&mut reader);
103                if out.is_empty() {
104                    return Err(VoidError::Network(format!("empty response fetching block {cid}")));
105                }
106                Ok(out)
107            }
108            IpfsBackend::Gateway { base } => {
109                let url = format!("{}/ipfs/{}", base.trim_end_matches('/'), cid);
110                let response = self
111                    .agent
112                    .get(&url)
113                    .call()
114                    .map_err(|e| VoidError::Network(extract_ureq_error(e)))?;
115                let mut reader = response.into_reader();
116                let out = read_response_bytes(&mut reader);
117                if out.is_empty() {
118                    return Err(VoidError::Network(format!("empty response fetching block {cid}")));
119                }
120                Ok(out)
121            }
122        }
123    }
124
125    fn push_raw_http(&self, data: &[u8]) -> Result<String> {
126        match &self.backend {
127            IpfsBackend::Kubo { api } => {
128                let url = format!(
129                    "{}/api/v0/block/put?format=raw&cid-version=1&mhtype=sha2-256",
130                    api.trim_end_matches('/')
131                );
132                let boundary = "----void-multipart-boundary";
133                let mut body = Vec::with_capacity(data.len() + 256);
134                body.extend_from_slice(format!("--{}\r\n", boundary).as_bytes());
135                body.extend_from_slice(
136                    b"Content-Disposition: form-data; name=\"file\"; filename=\"block\"\r\n",
137                );
138                body.extend_from_slice(b"Content-Type: application/octet-stream\r\n\r\n");
139                body.extend_from_slice(data);
140                body.extend_from_slice(b"\r\n");
141                body.extend_from_slice(format!("--{}--\r\n", boundary).as_bytes());
142
143                let response = self
144                    .agent
145                    .post(&url)
146                    .set(
147                        "Content-Type",
148                        &format!("multipart/form-data; boundary={}", boundary),
149                    )
150                    .send_bytes(&body)
151                    .map_err(|e| VoidError::Network(extract_ureq_error(e)))?;
152                let mut reader = response.into_reader();
153                let bytes = read_response_bytes(&mut reader);
154                let body = String::from_utf8(bytes)
155                    .map_err(|e| VoidError::Serialization(e.to_string()))?;
156                let parsed: BlockPutResponse = serde_json::from_str(&body)
157                    .map_err(|e| VoidError::Serialization(e.to_string()))?;
158                Ok(parsed.key)
159            }
160            IpfsBackend::Gateway { .. } => {
161                Err(VoidError::Network("gateway backend is read-only".into()))
162            }
163        }
164    }
165
166    fn has_raw(&self, cid: &str) -> Result<bool> {
167        match &self.backend {
168            IpfsBackend::Kubo { api } => {
169                let url = format!(
170                    "{}/api/v0/block/stat?arg={}",
171                    api.trim_end_matches('/'),
172                    cid
173                );
174                match self.agent.post(&url).call() {
175                    Ok(_) => Ok(true),
176                    Err(ureq::Error::Status(500, _)) => Ok(false),
177                    Err(e) => Err(VoidError::Network(extract_ureq_error(e))),
178                }
179            }
180            IpfsBackend::Gateway { base } => {
181                let url = format!("{}/ipfs/{}", base.trim_end_matches('/'), cid);
182                match self.agent.head(&url).call() {
183                    Ok(_) => Ok(true),
184                    Err(ureq::Error::Status(404, _)) => Ok(false),
185                    Err(e) => Err(VoidError::Network(extract_ureq_error(e))),
186                }
187            }
188        }
189    }
190
191    fn pin_raw(&self, cid: &str) -> Result<()> {
192        match &self.backend {
193            IpfsBackend::Kubo { api } => {
194                let url = format!("{}/api/v0/pin/add?arg={}", api.trim_end_matches('/'), cid);
195                self.agent
196                    .post(&url)
197                    .call()
198                    .map_err(|e| VoidError::Network(extract_ureq_error(e)))?;
199                Ok(())
200            }
201            IpfsBackend::Gateway { .. } => {
202                Err(VoidError::Network("gateway backend is read-only".into()))
203            }
204        }
205    }
206
207    // -----------------------------------------------------------------------
208    // IPFS-specific operations (not on RemoteStore trait)
209    // -----------------------------------------------------------------------
210
211    /// Add a directory recursively to IPFS via Kubo's `/api/v0/add` endpoint.
212    ///
213    /// Returns the root CID of the directory. Only works with Kubo backend.
214    pub fn add_directory(&self, dir: &Path) -> Result<String> {
215        match &self.backend {
216            IpfsBackend::Kubo { api } => {
217                let url = format!(
218                    "{}/api/v0/add?recursive=true&wrap-with-directory=true&pin=true&quieter=true",
219                    api.trim_end_matches('/')
220                );
221
222                let boundary = "----void-publish-boundary";
223                let mut body = Vec::new();
224                walk_dir_multipart(dir, dir, &mut body, boundary)?;
225                body.extend_from_slice(format!("--{}--\r\n", boundary).as_bytes());
226
227                let response = self
228                    .agent
229                    .post(&url)
230                    .set(
231                        "Content-Type",
232                        &format!("multipart/form-data; boundary={}", boundary),
233                    )
234                    .send_bytes(&body)
235                    .map_err(|e| VoidError::Network(extract_ureq_error(e)))?;
236
237                let mut reader = response.into_reader();
238                let bytes = read_response_bytes(&mut reader);
239                let response_body = String::from_utf8(bytes)
240                    .map_err(|e| VoidError::Serialization(e.to_string()))?;
241
242                let last_line = response_body
243                    .lines()
244                    .filter(|l| !l.is_empty())
245                    .last()
246                    .ok_or_else(|| VoidError::Network("empty response from IPFS add".into()))?;
247
248                let parsed: AddResponse = serde_json::from_str(last_line)
249                    .map_err(|e| VoidError::Serialization(e.to_string()))?;
250
251                Ok(parsed.hash)
252            }
253            IpfsBackend::Gateway { .. } => {
254                Err(VoidError::Network("gateway backend is read-only".into()))
255            }
256        }
257    }
258}
259
260// ---------------------------------------------------------------------------
261// RemoteStore implementation
262// ---------------------------------------------------------------------------
263
264impl RemoteStore for IpfsStore {
265    fn fetch_raw(&self, cid_obj: &VoidCid) -> Result<Vec<u8>> {
266        cid::validate(cid_obj)?;
267
268        let cid_str = cid_obj.to_string();
269        let data = self.fetch_raw_http(&cid_str)?;
270
271        let computed = cid::create(&data);
272        if computed != *cid_obj {
273            return Err(VoidError::Shard(format!(
274                "CID verification failed for {cid_str}: content hash does not match"
275            )));
276        }
277
278        Ok(data)
279    }
280
281    fn push_raw(&self, data: &[u8]) -> Result<VoidCid> {
282        let expected = cid::create(data);
283        let returned_str = self.push_raw_http(data)?;
284        let returned = cid::parse(&returned_str)?;
285
286        if returned != expected {
287            return Err(VoidError::IntegrityError {
288                expected: expected.to_string(),
289                actual: returned_str,
290            });
291        }
292
293        Ok(returned)
294    }
295
296    fn exists(&self, cid: &VoidCid) -> Result<bool> {
297        self.has_raw(&cid.to_string())
298    }
299
300    fn pin(&self, cid: &VoidCid) -> Result<()> {
301        self.pin_raw(&cid.to_string())
302    }
303}
304
305// ---------------------------------------------------------------------------
306// Helpers
307// ---------------------------------------------------------------------------
308
309#[derive(Deserialize)]
310struct AddResponse {
311    #[serde(rename = "Hash")]
312    hash: String,
313}
314
315fn walk_dir_multipart(
316    base: &Path,
317    current: &Path,
318    body: &mut Vec<u8>,
319    boundary: &str,
320) -> Result<()> {
321    let entries = std::fs::read_dir(current).map_err(VoidError::Io)?;
322
323    for entry in entries {
324        let entry = entry.map_err(VoidError::Io)?;
325        let path = entry.path();
326        let rel_path = path
327            .strip_prefix(base)
328            .unwrap_or(&path)
329            .to_string_lossy()
330            .replace('\\', "/");
331
332        if path.is_dir() {
333            walk_dir_multipart(base, &path, body, boundary)?;
334        } else {
335            let content = std::fs::read(&path).map_err(VoidError::Io)?;
336
337            body.extend_from_slice(format!("--{}\r\n", boundary).as_bytes());
338            body.extend_from_slice(
339                format!(
340                    "Content-Disposition: form-data; name=\"file\"; filename=\"{}\"\r\n",
341                    rel_path
342                )
343                .as_bytes(),
344            );
345            body.extend_from_slice(b"Content-Type: application/octet-stream\r\n");
346            body.extend_from_slice(
347                format!("Abspath: /{}\r\n", rel_path).as_bytes(),
348            );
349            body.extend_from_slice(b"\r\n");
350            body.extend_from_slice(&content);
351            body.extend_from_slice(b"\r\n");
352        }
353    }
354
355    Ok(())
356}
void_core/store/ipfs.rs

void_core/store/
ipfs.rs
