deaddrop_net/
control.rs

1//! Local control API. Bound to loopback (and a Unix socket when available).
2//! Not exposed on public interfaces.
3
4use deaddrop_core::store::{Store, unix_now};
5use deaddrop_core::{PROTOCOL_LABEL, Result};
6use serde::Serialize;
7use std::net::SocketAddr;
8use std::path::{Path, PathBuf};
9use std::sync::Arc;
10use std::sync::atomic::{AtomicBool, Ordering};
11use tokio::io::{AsyncReadExt, AsyncWriteExt};
12use tokio::net::TcpListener;
13
14pub const CONTROL_FILE: &str = "control.json";
15pub const DAEMON_FILE: &str = "daemon.json";
16
17#[derive(Debug, Clone, Serialize, serde::Deserialize)]
18pub struct ControlInfo {
19    pub listen: String,
20    pub token: String,
21    pub pid: u32,
22    pub started_at: u64,
23    #[serde(default)]
24    pub unix_socket: Option<String>,
25}
26
27pub fn control_path(root: &Path) -> PathBuf {
28    root.join(CONTROL_FILE)
29}
30
31pub fn daemon_path(root: &Path) -> PathBuf {
32    root.join(DAEMON_FILE)
33}
34
35pub fn write_info(root: &Path, info: &ControlInfo) -> Result<()> {
36    let bytes = serde_json::to_vec_pretty(info)
37        .map_err(|e| deaddrop_core::DdError::crypto(e.to_string()))?;
38    std::fs::write(control_path(root), &bytes)?;
39    std::fs::write(daemon_path(root), bytes)?;
40    Ok(())
41}
42
43pub fn read_info(root: &Path) -> Result<Option<ControlInfo>> {
44    let p = control_path(root);
45    if !p.exists() {
46        return Ok(None);
47    }
48    let bytes = std::fs::read(p)?;
49    let info: ControlInfo = serde_json::from_slice(&bytes)
50        .map_err(|e| deaddrop_core::DdError::invalid_frame(format!("control.json: {e}")))?;
51    Ok(Some(info))
52}
53
54pub fn clear(root: &Path) {
55    let _ = std::fs::remove_file(control_path(root));
56    let _ = std::fs::remove_file(daemon_path(root));
57    let _ = std::fs::remove_file(root.join("control.sock"));
58}
59
60pub fn random_token() -> String {
61    use std::time::{SystemTime, UNIX_EPOCH};
62    let n = SystemTime::now()
63        .duration_since(UNIX_EPOCH)
64        .map(|d| d.as_nanos())
65        .unwrap_or(1);
66    format!("{n:x}{:x}", std::process::id())
67}
68
69pub struct ControlServer {
70    pub info: ControlInfo,
71    pub stop: Arc<AtomicBool>,
72}
73
74impl ControlServer {
75    pub async fn bind(root: &Path, ddp_listen: SocketAddr) -> Result<Self> {
76        let listener = TcpListener::bind("127.0.0.1:0").await?;
77        let local = listener.local_addr()?;
78        let token = random_token();
79        #[cfg(unix)]
80        let unix_socket = {
81            let sock = root.join("control.sock");
82            let _ = std::fs::remove_file(&sock);
83            Some(sock.display().to_string())
84        };
85        #[cfg(not(unix))]
86        let unix_socket = None::<String>;
87        let info = ControlInfo {
88            listen: local.to_string(),
89            token: token.clone(),
90            pid: std::process::id(),
91            started_at: unix_now(),
92            unix_socket: unix_socket.clone(),
93        };
94        write_info(root, &info)?;
95        let stop = Arc::new(AtomicBool::new(false));
96        let stop_a = stop.clone();
97        let root_a = root.to_path_buf();
98        let token_a = token.clone();
99        tokio::spawn(async move {
100            loop {
101                if stop_a.load(Ordering::Relaxed) {
102                    break;
103                }
104                match listener.accept().await {
105                    Ok((stream, _)) => {
106                        let stop = stop_a.clone();
107                        let root = root_a.clone();
108                        let token = token_a.clone();
109                        tokio::spawn(async move {
110                            let _ = handle_client(stream, &root, &token, &stop).await;
111                        });
112                    }
113                    Err(_) => break,
114                }
115            }
116        });
117        #[cfg(unix)]
118        if let Some(path) = unix_socket {
119            let stop_b = stop.clone();
120            let root_b = root.to_path_buf();
121            let token_b = token;
122            tokio::spawn(async move {
123                if let Ok(ul) = tokio::net::UnixListener::bind(&path) {
124                    loop {
125                        if stop_b.load(Ordering::Relaxed) {
126                            break;
127                        }
128                        if let Ok((stream, _)) = ul.accept().await {
129                            let stop = stop_b.clone();
130                            let root = root_b.clone();
131                            let token = token_b.clone();
132                            tokio::spawn(async move {
133                                let _ = handle_unix(stream, &root, &token, &stop).await;
134                            });
135                        }
136                    }
137                }
138            });
139        }
140        let _ = ddp_listen;
141        Ok(Self { info, stop })
142    }
143}
144
145#[cfg(unix)]
146async fn handle_unix(
147    stream: tokio::net::UnixStream,
148    root: &Path,
149    token: &str,
150    stop: &Arc<AtomicBool>,
151) -> Result<()> {
152    handle_generic(stream, root, token, stop).await
153}
154
155async fn handle_client(
156    stream: tokio::net::TcpStream,
157    root: &Path,
158    token: &str,
159    stop: &Arc<AtomicBool>,
160) -> Result<()> {
161    handle_generic(stream, root, token, stop).await
162}
163
164async fn handle_generic<S>(
165    mut stream: S,
166    root: &Path,
167    token: &str,
168    stop: &Arc<AtomicBool>,
169) -> Result<()>
170where
171    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
172{
173    let mut buf = vec![0u8; 8192];
174    let n = stream.read(&mut buf).await?;
175    let req = String::from_utf8_lossy(&buf[..n]);
176    let authed = req.contains(&format!("Bearer {token}")) || req.lines().any(|l| l.trim() == token);
177    if !authed {
178        let body = b"HTTP/1.0 401 Unauthorized\r\nContent-Length: 12\r\n\r\nunauthorized";
179        let _ = stream.write_all(body).await;
180        return Ok(());
181    }
182    let path = req
183        .lines()
184        .next()
185        .unwrap_or("")
186        .split_whitespace()
187        .nth(1)
188        .unwrap_or("/");
189    if path == "/v1/shutdown" || path.starts_with("/v1/shutdown") {
190        stop.store(true, Ordering::Relaxed);
191        let body = br#"{"ok":true}"#;
192        write_http(&mut stream, 200, body).await?;
193        return Ok(());
194    }
195    let payload = match status_json(root, path) {
196        Ok(v) => v,
197        Err(e) => serde_json::json!({"error": e.to_string()}),
198    };
199    let bytes = serde_json::to_vec(&payload).unwrap_or_else(|_| b"{}".to_vec());
200    write_http(&mut stream, 200, &bytes).await?;
201    Ok(())
202}
203
204async fn write_http<S: tokio::io::AsyncWrite + Unpin>(
205    stream: &mut S,
206    code: u16,
207    body: &[u8],
208) -> Result<()> {
209    let head = format!(
210        "HTTP/1.0 {code} OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n",
211        body.len()
212    );
213    stream.write_all(head.as_bytes()).await?;
214    stream.write_all(body).await?;
215    Ok(())
216}
217
218fn status_json(root: &Path, path: &str) -> Result<serde_json::Value> {
219    let store = Store::open(root, Default::default())?;
220    let now = unix_now();
221    match path {
222        "/v1/status" => {
223            let id = store.load_identity()?;
224            let s = store.stats()?;
225            Ok(serde_json::json!({
226                "protocol": PROTOCOL_LABEL,
227                "identity": id.peer_id.to_string(),
228                "objects": s.objects,
229                "storage": s.physical_size,
230            }))
231        }
232        "/v1/peers" => {
233            let book = store.load_contacts()?;
234            let list: Vec<_> = book
235                .all()
236                .map(|(id, c)| {
237                    serde_json::json!({
238                        "id": id.to_string(),
239                        "name": c.card.name,
240                        "trust": format!("{:?}", c.trust),
241                    })
242                })
243                .collect();
244            Ok(serde_json::json!({ "peers": list }))
245        }
246        "/v1/drops" => {
247            let ids: Vec<_> = store
248                .inventory(now)?
249                .into_iter()
250                .map(|i| i.to_string())
251                .collect();
252            Ok(serde_json::json!({ "drops": ids }))
253        }
254        "/v1/routes" => Ok(serde_json::json!({
255            "note": "route table is computed per Drop; use dd route explain"
256        })),
257        "/v1/transfers" => Ok(transfers_json(&store, now)?),
258        "/v1/spaces" => {
259            let spaces: Vec<_> = store
260                .list_spaces()?
261                .into_iter()
262                .map(|s| serde_json::json!({"name": s.name}))
263                .collect();
264            Ok(serde_json::json!({ "spaces": spaces }))
265        }
266        "/v1/channels" => {
267            let ch: Vec<_> = store
268                .list_channels()?
269                .into_iter()
270                .map(|(n, f, p)| serde_json::json!({"name": n, "filter": f, "policy": p}))
271                .collect();
272            Ok(serde_json::json!({ "channels": ch }))
273        }
274        "/v1/events" => Ok(serde_json::json!({
275            "note": "subscribe via SDK DeadDrop::events(); HTTP is a snapshot API"
276        })),
277        _ => Ok(serde_json::json!({ "error": "not found", "path": path })),
278    }
279}
280
281pub fn transfers_json(store: &Store, now: u64) -> Result<serde_json::Value> {
282    let mut active = 0u64;
283    let mut queued = 0u64;
284    let mut completed = 0u64;
285    let mut items = Vec::new();
286    for id in store.inventory(now)? {
287        let complete = store.complete(&id).unwrap_or(false);
288        let mask = store.present_mask(&id).unwrap_or_default();
289        let present = mask.iter().filter(|x| **x).count();
290        let total = mask.len().max(1);
291        if complete {
292            completed += 1;
293        } else if present > 0 {
294            active += 1;
295        } else {
296            queued += 1;
297        }
298        items.push(serde_json::json!({
299            "id": id.to_string(),
300            "chunks": format!("{present}/{total}"),
301            "complete": complete,
302        }));
303    }
304    Ok(serde_json::json!({
305        "active": active,
306        "queued": queued,
307        "completed": completed,
308        "items": items,
309    }))
310}
311
312pub async fn query(root: &Path, path: &str) -> Result<serde_json::Value> {
313    let Some(info) = read_info(root)? else {
314        return Err(deaddrop_core::DdError::invalid_frame("daemon not running"));
315    };
316    let addr: SocketAddr = info
317        .listen
318        .parse()
319        .map_err(|_| deaddrop_core::DdError::invalid_frame("control listen"))?;
320    let mut s = tokio::net::TcpStream::connect(addr).await?;
321    let req = format!(
322        "GET {path} HTTP/1.0\r\nAuthorization: Bearer {}\r\n\r\n",
323        info.token
324    );
325    s.write_all(req.as_bytes()).await?;
326    let mut buf = Vec::new();
327    s.read_to_end(&mut buf).await?;
328    let text = String::from_utf8_lossy(&buf);
329    let body = text.split("\r\n\r\n").nth(1).unwrap_or("{}");
330    Ok(serde_json::from_str(body).unwrap_or_else(|_| serde_json::json!({"raw": body})))
331}
deaddrop_net/control.rs

deaddrop_net/
control.rs
