reddb_engine/query/local_database/
service.rs

1//! Private local workspace service for managed desktop hosts. Never a user SQL
2//! endpoint: the only operations are metadata batches and explicit Vault access.
3use super::{vault::LocalVaultKey, LocalDatabase};
4use reddb_file::vfs::StdFile;
5use reddb_file::{EmbeddedRdbWriterLock, OpenMode, StdVfs, Vfs, VfsFile};
6use reddb_types::json::Value as Json;
7use reddb_types::Value;
8use std::io::{BufRead, Read, Write};
9use std::net::{TcpListener, TcpStream};
10use std::path::Path;
11use std::sync::{
12    atomic::{AtomicBool, Ordering},
13    Arc,
14};
15use std::time::Duration;
16use zeroize::Zeroizing;
17
18type Result<T> = std::result::Result<T, String>;
19const REQUEST_BYTES_MAX: usize = 2 * 1024 * 1024;
20
21/// `workspace-store --path <private.rdb>` binds an ephemeral loopback port.
22/// A single bounded JSON bootstrap line arrives on stdin. The host keeps stdin
23/// open as a lifeline and reads the private port/token handshake from stdout.
24pub fn run(arguments: &[String]) -> Result<()> {
25    let [flag, path] = arguments else {
26        return Err("expected --path <private.rdb>".into());
27    };
28    if flag != "--path" {
29        return Err("expected --path <private.rdb>".into());
30    }
31    let mut bootstrap = Zeroizing::new(String::new());
32    std::io::stdin()
33        .lock()
34        .take(8192)
35        .read_line(&mut bootstrap)
36        .map_err(error)?;
37    if !bootstrap.ends_with('\n') {
38        return Err("invalid bootstrap frame".into());
39    }
40    let frame: Json = reddb_types::serde_json::from_str(&bootstrap).map_err(error)?;
41    drop(bootstrap);
42    let password = match frame {
43        Json::Object(mut object) => match object.remove("password") {
44            Some(Json::String(password)) => Some(Zeroizing::new(password)),
45            Some(Json::Null) | None => None,
46            _ => return Err("invalid opening credential".into()),
47        },
48        _ => return Err("invalid bootstrap frame".into()),
49    };
50    let path = Path::new(path);
51    let _owner = EmbeddedRdbWriterLock::acquire(path).map_err(error)?;
52    let existed = path.exists();
53    let mut mode = if existed {
54        OpenMode::create_keep()
55    } else {
56        OpenMode::create_new()
57    };
58    mode.read = true;
59    let file = StdVfs.open(path, mode).map_err(error)?;
60    let database = if existed {
61        LocalDatabase::open(file, path)
62    } else {
63        LocalDatabase::create(file, path, 0)
64    }
65    .map_err(error)?;
66    let mut service = Service {
67        database,
68        key: None,
69    };
70    service.database.query("CREATE TABLE IF NOT EXISTS workspace_metadata (entry_key TEXT PRIMARY KEY, entry_value TEXT NOT NULL)", 0).map_err(error)?;
71    if let Some(password) = &password {
72        // A failed unlock leaves metadata accessible and never recreates a Vault.
73        let create = !service.database.vault_initialized().map_err(error)?;
74        service.key = service.database.vault_unlock(password, create).ok();
75    }
76    drop(password);
77    let listener = TcpListener::bind("127.0.0.1:0").map_err(error)?;
78    listener.set_nonblocking(true).map_err(error)?;
79    let token = private_token()?;
80    println!("{}", reddb_types::json!({ "protocol": 1, "port": listener.local_addr().map_err(error)?.port(), "token": token.clone() }).to_string_compact());
81    std::io::stdout().flush().map_err(error)?;
82    let alive = Arc::new(AtomicBool::new(true));
83    let lifeline = alive.clone();
84    std::thread::spawn(move || {
85        let mut sink = [0u8; 1];
86        while std::io::stdin()
87            .read(&mut sink)
88            .is_ok_and(|count| count > 0)
89        {}
90        lifeline.store(false, Ordering::Release);
91    });
92    while alive.load(Ordering::Acquire) {
93        match listener.accept() {
94            Ok((mut stream, _)) => {
95                let request = read_request(&mut stream, &token);
96                let (response, stop) = match request {
97                    Ok(request) => {
98                        let stop = request.get("method").and_then(Json::as_str) == Some("shutdown");
99                        let result = service.execute(&request);
100                        (
101                            match result {
102                                Ok(value) => reddb_types::json!({ "ok": true, "result": value }),
103                                Err(message) => {
104                                    reddb_types::json!({ "ok": false, "error": message })
105                                }
106                            },
107                            stop,
108                        )
109                    }
110                    Err(_) => {
111                        let _ = stream.write_all(b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
112                        continue;
113                    }
114                };
115                let body = response.to_string_compact();
116                let _ = write!(stream, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nCache-Control: no-store\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body);
117                if stop {
118                    break;
119                }
120            }
121            Err(failure) if failure.kind() == std::io::ErrorKind::WouldBlock => {
122                std::thread::sleep(Duration::from_millis(20))
123            }
124            Err(failure) => return Err(error(failure)),
125        }
126    }
127    service.database.into_inner().sync_all().map_err(error)
128}
129
130fn error(error: impl std::fmt::Display) -> String {
131    error.to_string()
132}
133
134fn private_token() -> Result<String> {
135    let mut bytes = [0u8; 32];
136    // The engine owns browser/native Vault entropy too; use the same provider.
137    reddb_crypto::os_random::fill_bytes(&mut bytes).map_err(error)?;
138    Ok(bytes.iter().map(|byte| format!("{byte:02x}")).collect())
139}
140
141fn read_request(stream: &mut TcpStream, token: &str) -> Result<Json> {
142    stream
143        .set_read_timeout(Some(Duration::from_secs(5)))
144        .map_err(error)?;
145    stream
146        .set_write_timeout(Some(Duration::from_secs(5)))
147        .map_err(error)?;
148    let mut header = Vec::new();
149    while !header.ends_with(b"\r\n\r\n") && header.len() < 8192 {
150        let mut byte = [0];
151        stream.read_exact(&mut byte).map_err(error)?;
152        header.push(byte[0]);
153    }
154    let header = std::str::from_utf8(&header).map_err(error)?;
155    let mut lines = header.split("\r\n");
156    if lines.next() != Some("POST /workspace HTTP/1.1") {
157        return Err("invalid endpoint".into());
158    }
159    let mut length = None;
160    let mut authenticated = false;
161    for line in lines.filter(|line| !line.is_empty()) {
162        let (name, value) = line.split_once(':').ok_or("invalid header")?;
163        let name = name.to_ascii_lowercase();
164        let value = value.trim();
165        match name.as_str() {
166            "origin" | "transfer-encoding" => return Err("browser requests are refused".into()),
167            "authorization" => {
168                let expected = format!("Bearer {token}");
169                authenticated = value.len() == expected.len()
170                    && value
171                        .bytes()
172                        .zip(expected.bytes())
173                        .fold(0u8, |difference, (left, right)| difference | (left ^ right))
174                        == 0;
175            }
176            "content-length" => {
177                if length.is_some() {
178                    return Err("duplicate length".into());
179                }
180                length = Some(value.parse::<usize>().map_err(error)?);
181            }
182            _ => {}
183        }
184    }
185    let length = length
186        .filter(|length| *length > 0 && *length <= REQUEST_BYTES_MAX)
187        .ok_or("invalid request size")?;
188    if !authenticated {
189        return Err("unauthenticated".into());
190    }
191    let mut body = Zeroizing::new(vec![0; length]);
192    stream.read_exact(&mut body).map_err(error)?;
193    reddb_types::serde_json::from_slice(&body).map_err(error)
194}
195
196struct Service {
197    database: LocalDatabase<StdFile>,
198    key: Option<LocalVaultKey>,
199}
200
201impl Service {
202    fn execute(&mut self, request: &Json) -> Result<Json> {
203        let params = request.get("params").unwrap_or(&Json::Null);
204        let name = || {
205            params
206                .get("name")
207                .and_then(Json::as_str)
208                .ok_or_else(|| "missing Vault key".to_string())
209        };
210        match request
211            .get("method")
212            .and_then(Json::as_str)
213            .ok_or("missing method")?
214        {
215            "read" => {
216                let result = self
217                    .database
218                    .query("SELECT entry_key, entry_value FROM workspace_metadata", 0)
219                    .map_err(error)?;
220                let mut object = reddb_types::json::Map::new();
221                for row in result.rows {
222                    let key = row[0].as_text().ok_or("invalid metadata key")?;
223                    let value = row[1].as_text().ok_or("invalid metadata value")?;
224                    object.insert(key.into(), Json::String(value.into()));
225                }
226                Ok(Json::Object(object))
227            }
228            "commit" => self.commit(params),
229            "vaultStatus" => Ok(
230                reddb_types::json!({ "initialized": self.database.vault_initialized().map_err(error)?, "locked": self.key.is_none() }),
231            ),
232            "vaultUnlock" => {
233                self.key = None;
234                let password = params
235                    .get("password")
236                    .and_then(Json::as_str)
237                    .ok_or("missing opening credential")?;
238                let create = params
239                    .get("create")
240                    .and_then(Json::as_bool)
241                    .unwrap_or(false);
242                self.key = Some(
243                    self.database
244                        .vault_unlock(password, create)
245                        .map_err(error)?,
246                );
247                Ok(Json::Null)
248            }
249            "vaultLock" => {
250                self.key = None;
251                Ok(Json::Null)
252            }
253            "vaultKeys" => Ok(Json::Array(
254                self.database
255                    .vault_keys(self.key.as_ref().ok_or("VAULT_LOCKED")?)
256                    .map_err(error)?
257                    .into_iter()
258                    .map(Json::String)
259                    .collect(),
260            )),
261            "vaultReveal" => Ok(self
262                .database
263                .vault_reveal(self.key.as_ref().ok_or("VAULT_LOCKED")?, name()?)
264                .map_err(error)?
265                .map(Json::String)
266                .unwrap_or(Json::Null)),
267            "shutdown" => Ok(Json::Null),
268            _ => Err("unsupported workspace operation".into()),
269        }
270    }
271
272    fn commit(&mut self, params: &Json) -> Result<Json> {
273        self.database.query("BEGIN", 0).map_err(error)?;
274        let result = self.commit_inner(params);
275        if result.is_err() {
276            let _ = self.database.query("ROLLBACK", 0);
277        }
278        result
279    }
280
281    fn commit_inner(&mut self, params: &Json) -> Result<Json> {
282        let metadata = params
283            .get("metadata")
284            .and_then(Json::as_object)
285            .ok_or("missing metadata batch")?;
286        let if_absent = params
287            .get("ifAbsent")
288            .and_then(Json::as_bool)
289            .unwrap_or(false);
290        for (key, value) in metadata {
291            if key.len() > 2048 {
292                return Err("metadata key too long".into());
293            }
294            if if_absent
295                && !self
296                    .database
297                    .query_with_params(
298                        "SELECT entry_key FROM workspace_metadata WHERE entry_key = $1",
299                        &[Value::text(key.as_str())],
300                        0,
301                    )
302                    .map_err(error)?
303                    .rows
304                    .is_empty()
305            {
306                continue;
307            }
308            self.database
309                .query_with_params(
310                    "DELETE FROM workspace_metadata WHERE entry_key = $1",
311                    &[Value::text(key.as_str())],
312                    0,
313                )
314                .map_err(error)?;
315            if !matches!(value, Json::Null) {
316                let value = value.as_str().ok_or("invalid metadata value")?;
317                self.database
318                    .query_with_params(
319                        "INSERT INTO workspace_metadata VALUES ($1, $2)",
320                        &[Value::text(key.as_str()), Value::text(value)],
321                        0,
322                    )
323                    .map_err(error)?;
324            }
325        }
326        if let Some(secrets) = params.get("secrets").and_then(Json::as_object) {
327            for (name, value) in secrets {
328                if if_absent
329                    && self
330                        .database
331                        .vault_reveal(self.key.as_ref().ok_or("VAULT_LOCKED")?, name)
332                        .map_err(error)?
333                        .is_some()
334                {
335                    continue;
336                }
337                let value = if matches!(value, Json::Null) {
338                    None
339                } else {
340                    Some(value.as_str().ok_or("invalid secret value")?)
341                };
342                self.database
343                    .vault_put(self.key.as_ref().ok_or("VAULT_LOCKED")?, name, value)
344                    .map_err(error)?;
345            }
346        }
347        self.database
348            .query_with_params(
349                "DELETE FROM workspace_metadata WHERE entry_key = $1",
350                &[Value::text("__workspace_migration_v1")],
351                0,
352            )
353            .map_err(error)?;
354        self.database
355            .query_with_params(
356                "INSERT INTO workspace_metadata VALUES ($1, $2)",
357                &[
358                    Value::text("__workspace_migration_v1"),
359                    Value::text("complete"),
360                ],
361                0,
362            )
363            .map_err(error)?;
364        self.database.query("COMMIT", 0).map_err(error)?;
365        Ok(Json::Null)
366    }
367}
reddb_engine/query/local_database/service.rs

reddb_engine/query/local_database/
service.rs
