fud/
rpc.rs

1/* This file is part of DarkFi (https://dark.fi)
2 *
3 * Copyright (C) 2020-2025 Dyne.org foundation
4 *
5 * This program is free software: you can redistribute it and/or modify
6 * it under the terms of the GNU Affero General Public License as
7 * published by the Free Software Foundation, either version 3 of the
8 * License, or (at your option) any later version.
9 *
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13 * GNU Affero General Public License for more details.
14 *
15 * You should have received a copy of the GNU Affero General Public License
16 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
17 */
18
19use async_trait::async_trait;
20use smol::lock::{Mutex, MutexGuard};
21use std::{
22    collections::{HashMap, HashSet},
23    path::PathBuf,
24    sync::Arc,
25};
26use tinyjson::JsonValue;
27use tracing::error;
28
29use darkfi::{
30    dht::DhtNode,
31    geode::hash_to_string,
32    net::P2pPtr,
33    rpc::{
34        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult, JsonSubscriber},
35        p2p_method::HandlerP2p,
36        server::RequestHandler,
37    },
38    system::StoppableTaskPtr,
39    util::path::expand_path,
40    Result,
41};
42
43use crate::{util::FileSelection, Fud};
44
45pub struct JsonRpcInterface {
46    fud: Arc<Fud>,
47    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
48    dnet_sub: JsonSubscriber,
49    event_sub: JsonSubscriber,
50}
51
52#[async_trait]
53impl RequestHandler<()> for JsonRpcInterface {
54    async fn handle_request(&self, req: JsonRequest) -> JsonResult {
55        return match req.method.as_str() {
56            "ping" => self.pong(req.id, req.params).await,
57
58            "put" => self.put(req.id, req.params).await,
59            "get" => self.get(req.id, req.params).await,
60            "subscribe" => self.subscribe(req.id, req.params).await,
61            "remove" => self.remove(req.id, req.params).await,
62            "list_resources" => self.list_resources(req.id, req.params).await,
63            "list_buckets" => self.list_buckets(req.id, req.params).await,
64            "list_seeders" => self.list_seeders(req.id, req.params).await,
65            "verify" => self.verify(req.id, req.params).await,
66            "lookup" => self.lookup(req.id, req.params).await,
67
68            "dnet.switch" => self.dnet_switch(req.id, req.params).await,
69            "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
70            "p2p.get_info" => self.p2p_get_info(req.id, req.params).await,
71            _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
72        }
73    }
74
75    async fn connections_mut(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
76        self.rpc_connections.lock().await
77    }
78}
79
80impl HandlerP2p for JsonRpcInterface {
81    fn p2p(&self) -> P2pPtr {
82        self.fud.p2p.clone()
83    }
84}
85
86/// Fud RPC methods
87impl JsonRpcInterface {
88    pub fn new(fud: Arc<Fud>, dnet_sub: JsonSubscriber, event_sub: JsonSubscriber) -> Self {
89        Self { fud, rpc_connections: Mutex::new(HashSet::new()), dnet_sub, event_sub }
90    }
91
92    // RPCAPI:
93    // Put a file/directory onto the network. Takes a local filesystem path as a parameter.
94    // Returns the resource hash that serves as a pointer to the file/directory.
95    //
96    // --> {"jsonrpc": "2.0", "method": "put", "params": ["/foo.txt"], "id": 42}
97    // <-- {"jsonrpc": "2.0", "result: "df4...3db7", "id": 42}
98    async fn put(&self, id: u16, params: JsonValue) -> JsonResult {
99        let params = params.get::<Vec<JsonValue>>().unwrap();
100        if params.len() != 1 || !params[0].is_string() {
101            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
102        }
103
104        let path = params[0].get::<String>().unwrap();
105        let path = match expand_path(path.as_str()) {
106            Ok(v) => v,
107            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
108        };
109
110        // A valid path was passed. Let's see if we can read it, and if so,
111        // add it to Geode.
112        let res = self.fud.put(&path).await;
113        if let Err(e) = res {
114            return JsonError::new(ErrorCode::InternalError, Some(format!("{e}")), id).into()
115        }
116
117        JsonResponse::new(JsonValue::String(path.to_string_lossy().to_string()), id).into()
118    }
119
120    // RPCAPI:
121    // Fetch a resource from the network. Takes a hash, path (absolute or relative), and an
122    // optional list of file paths (only used for directories) as parameters.
123    // Returns the path where the resource will be located once downloaded.
124    //
125    // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd", "~/myfile.jpg", null], "id": 42}
126    // <-- {"jsonrpc": "2.0", "result": "/home/user/myfile.jpg", "id": 42}
127    async fn get(&self, id: u16, params: JsonValue) -> JsonResult {
128        let params = params.get::<Vec<JsonValue>>().unwrap();
129        if params.len() != 3 || !params[0].is_string() || !params[1].is_string() {
130            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
131        }
132
133        let mut hash_buf = vec![];
134        match bs58::decode(params[0].get::<String>().unwrap().as_str()).onto(&mut hash_buf) {
135            Ok(_) => {}
136            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
137        }
138
139        if hash_buf.len() != 32 {
140            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
141        }
142
143        let mut hash_buf_arr = [0u8; 32];
144        hash_buf_arr.copy_from_slice(&hash_buf);
145
146        let hash = blake3::Hash::from_bytes(hash_buf_arr);
147        let hash_str = hash_to_string(&hash);
148
149        let path = match params[1].get::<String>() {
150            Some(path) => match path.is_empty() {
151                true => match self.fud.hash_to_path(&hash).ok().flatten() {
152                    Some(path) => path,
153                    None => self.fud.downloads_path.join(&hash_str),
154                },
155                false => match PathBuf::from(path).is_absolute() {
156                    true => PathBuf::from(path),
157                    false => self.fud.downloads_path.join(path),
158                },
159            },
160            None => self.fud.downloads_path.join(&hash_str),
161        };
162
163        let files: FileSelection = match &params[2] {
164            JsonValue::Array(files) => files
165                .iter()
166                .filter_map(|v| {
167                    if let JsonValue::String(file) = v {
168                        Some(PathBuf::from(file.clone()))
169                    } else {
170                        None
171                    }
172                })
173                .collect(),
174            JsonValue::Null => FileSelection::All,
175            _ => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
176        };
177
178        // Start downloading the resource
179        if let Err(e) = self.fud.get(&hash, &path, files).await {
180            return JsonError::new(ErrorCode::InternalError, Some(e.to_string()), id).into()
181        }
182
183        JsonResponse::new(JsonValue::String(path.to_string_lossy().to_string()), id).into()
184    }
185
186    // RPCAPI:
187    // Subscribe to fud events.
188    //
189    // --> {"jsonrpc": "2.0", "method": "get", "params": [], "id": 42}
190    // <-- {"jsonrpc": "2.0", "result": `event`, "id": 42}
191    async fn subscribe(&self, _id: u16, _params: JsonValue) -> JsonResult {
192        self.event_sub.clone().into()
193    }
194
195    // RPCAPI:
196    // Activate or deactivate dnet in the P2P stack.
197    // By sending `true`, dnet will be activated, and by sending `false` dnet
198    // will be deactivated. Returns `true` on success.
199    //
200    // --> {"jsonrpc": "2.0", "method": "dnet_switch", "params": [true], "id": 42}
201    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
202    async fn dnet_switch(&self, id: u16, params: JsonValue) -> JsonResult {
203        let params = params.get::<Vec<JsonValue>>().unwrap();
204        if params.len() != 1 || !params[0].is_bool() {
205            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
206        }
207
208        let switch = params[0].get::<bool>().unwrap();
209
210        if *switch {
211            self.fud.p2p.dnet_enable();
212        } else {
213            self.fud.p2p.dnet_disable();
214        }
215
216        JsonResponse::new(JsonValue::Boolean(true), id).into()
217    }
218
219    // RPCAPI:
220    // Initializes a subscription to p2p dnet events.
221    // Once a subscription is established, `fud` will send JSON-RPC notifications of
222    // new network events to the subscriber.
223    //
224    // --> {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [], "id": 1}
225    // <-- {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [`event`]}
226    pub async fn dnet_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
227        let params = params.get::<Vec<JsonValue>>().unwrap();
228        if !params.is_empty() {
229            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
230        }
231
232        self.dnet_sub.clone().into()
233    }
234
235    // RPCAPI:
236    // Returns resources.
237    //
238    // --> {"jsonrpc": "2.0", "method": "list_buckets", "params": [], "id": 1}
239    // <-- {"jsonrpc": "2.0", "result": [[["abcdef", ["tcp://127.0.0.1:13337"]]]], "id": 1}
240    pub async fn list_resources(&self, id: u16, params: JsonValue) -> JsonResult {
241        let params = params.get::<Vec<JsonValue>>().unwrap();
242        if !params.is_empty() {
243            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
244        }
245
246        let resources_read = self.fud.resources.read().await;
247        let mut resources: Vec<JsonValue> = vec![];
248        for (_, resource) in resources_read.iter() {
249            resources.push(resource.clone().into());
250        }
251
252        JsonResponse::new(JsonValue::Array(resources), id).into()
253    }
254
255    // RPCAPI:
256    // Returns the current buckets.
257    //
258    // --> {"jsonrpc": "2.0", "method": "list_buckets", "params": [], "id": 1}
259    // <-- {"jsonrpc": "2.0", "result": [["abcdef", ["tcp://127.0.0.1:13337"]]], "id": 1}
260    pub async fn list_buckets(&self, id: u16, params: JsonValue) -> JsonResult {
261        let params = params.get::<Vec<JsonValue>>().unwrap();
262        if !params.is_empty() {
263            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
264        }
265        let mut buckets = vec![];
266        for bucket in self.fud.dht.buckets.read().await.iter() {
267            let mut nodes = vec![];
268            for node in bucket.nodes.clone() {
269                let mut addresses = vec![];
270                for addr in &node.addresses {
271                    addresses.push(JsonValue::String(addr.to_string()));
272                }
273                nodes.push(JsonValue::Array(vec![
274                    JsonValue::String(hash_to_string(&node.id())),
275                    JsonValue::Array(addresses),
276                ]));
277            }
278            buckets.push(JsonValue::Array(nodes));
279        }
280
281        JsonResponse::new(JsonValue::Array(buckets), id).into()
282    }
283
284    // RPCAPI:
285    // Returns the content of the seeders router.
286    //
287    // --> {"jsonrpc": "2.0", "method": "list_seeders", "params": [], "id": 1}
288    // <-- {"jsonrpc": "2.0", "result": {"seeders": {"abcdefileid": [["abcdef", ["tcp://127.0.0.1:13337"]]]}}, "id": 1}
289    pub async fn list_seeders(&self, id: u16, params: JsonValue) -> JsonResult {
290        let params = params.get::<Vec<JsonValue>>().unwrap();
291        if !params.is_empty() {
292            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
293        }
294        let mut seeders_table: HashMap<String, JsonValue> = HashMap::new();
295        for (hash, items) in self.fud.dht.hash_table.read().await.iter() {
296            let mut nodes = vec![];
297            for item in items {
298                let mut addresses = vec![];
299                for addr in &item.node.addresses {
300                    addresses.push(JsonValue::String(addr.to_string()));
301                }
302                nodes.push(JsonValue::Array(vec![
303                    JsonValue::String(hash_to_string(&item.node.id())),
304                    JsonValue::Array(addresses),
305                ]));
306            }
307            seeders_table.insert(hash_to_string(hash), JsonValue::Array(nodes));
308        }
309        let mut res: HashMap<String, JsonValue> = HashMap::new();
310        res.insert("seeders".to_string(), JsonValue::Object(seeders_table));
311
312        JsonResponse::new(JsonValue::Object(res), id).into()
313    }
314
315    // RPCAPI:
316    // Removes a resource.
317    //
318    // --> {"jsonrpc": "2.0", "method": "remove", "params": ["1211...abfd"], "id": 1}
319    // <-- {"jsonrpc": "2.0", "result": [], "id": 1}
320    pub async fn remove(&self, id: u16, params: JsonValue) -> JsonResult {
321        let params = params.get::<Vec<JsonValue>>().unwrap();
322        if params.len() != 1 || !params[0].is_string() {
323            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
324        }
325        let mut hash_buf = [0u8; 32];
326        match bs58::decode(params[0].get::<String>().unwrap().as_str()).onto(&mut hash_buf) {
327            Ok(_) => {}
328            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
329        }
330
331        self.fud.remove(&blake3::Hash::from_bytes(hash_buf)).await;
332
333        JsonResponse::new(JsonValue::Array(vec![]), id).into()
334    }
335
336    // RPCAPI:
337    // Verifies local files. Takes a list of file hashes as parameters.
338    // An empty list means all known files.
339    // Returns the path where the file will be located once downloaded.
340    //
341    // --> {"jsonrpc": "2.0", "method": "verify", "params": ["1211...abfd"], "id": 42}
342    // <-- {"jsonrpc": "2.0", "result": [], "id": 1}
343    async fn verify(&self, id: u16, params: JsonValue) -> JsonResult {
344        let params = params.get::<Vec<JsonValue>>().unwrap();
345        if !params.iter().all(|param| param.is_string()) {
346            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
347        }
348        let hashes = if params.is_empty() {
349            None
350        } else {
351            let hashes_str: Vec<String> =
352                params.iter().map(|param| param.get::<String>().unwrap().clone()).collect();
353            let hashes: Result<Vec<blake3::Hash>> = hashes_str
354                .into_iter()
355                .map(|hash_str| {
356                    let mut buf = [0u8; 32];
357                    bs58::decode(hash_str).onto(&mut buf)?;
358                    Ok(blake3::Hash::from_bytes(buf))
359                })
360                .collect();
361            if hashes.is_err() {
362                return JsonError::new(ErrorCode::InvalidParams, None, id).into();
363            }
364            Some(hashes.unwrap())
365        };
366
367        if let Err(e) = self.fud.verify_resources(hashes).await {
368            error!(target: "fud::verify()", "Could not verify resources: {e}");
369            return JsonError::new(ErrorCode::InternalError, None, id).into();
370        }
371
372        JsonResponse::new(JsonValue::Array(vec![]), id).into()
373    }
374
375    // RPCAPI:
376    // Lookup a resource's seeders.
377    //
378    // --> {"jsonrpc": "2.0", "method": "lookup", "params": ["1211...abfd"], "id": 1}
379    // <-- {"jsonrpc": "2.0", "result": {"seeders": {"abcdefileid": [["abcdef", ["tcp://127.0.0.1:13337"]]]}}, "id": 1}
380    pub async fn lookup(&self, id: u16, params: JsonValue) -> JsonResult {
381        let params = params.get::<Vec<JsonValue>>().unwrap();
382        if params.len() != 1 || !params[0].is_string() {
383            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
384        }
385        let mut hash_buf = [0u8; 32];
386        match bs58::decode(params[0].get::<String>().unwrap().as_str()).onto(&mut hash_buf) {
387            Ok(_) => {}
388            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
389        }
390
391        let _ = self.fud.lookup_tx.send(blake3::Hash::from_bytes(hash_buf)).await;
392
393        JsonResponse::new(JsonValue::Array(vec![]), id).into()
394    }
395}
fud/rpc.rs

fud/
rpc.rs
