CH-J Tools通过 SSH 管理服务器
菜单
已发布的源代码

CH-J Tools

浏览特定应用版本的目录和文件。

下载源代码 ZIP
CH-J / Server Manager 1704 个文件
0.1.2-beta 添加时间: · 构建时间:
keysync.rs 添加时间: 未记录日期 38.7 KB · 706 行
代码 下载文件
1//! KeySyncManager: persistent publication/revocation queue and verified server state for keys.ch-j.de.
2//!
3//! `keysync.json` in the data directory (0600, atomic writes) holds only public material:
4//! - `keys`: public V4/V6 certificates mirrored from the Keychain plus the state *verified* on the server;
5//! - `jobs`: what still has to be sent (publish, revoke, check), attempts, next run, last error.
6//! Queue state and server state are separate records: an HTTP 200 never counts as success, only a
7//! certificate fetched back from the server that shows the change. The Standby Helper reads the same file
8//! to finish pending revocations while the application is closed; private keys never enter it.
9//!
10//! Two processes share the file (owner 2026-11-05: Core syncs while the app runs, the helper only when it
11//! is closed). `keysync.owner.lock` is held by whoever may process jobs: Core for its whole lifetime, the
12//! helper only for one pass when it can take it. Every write re-reads the file under `keysync.json.lock`,
13//! so a change made by the other process is never overwritten.
15use crate::core::error::{CoreError, CoreResult};
16use parking_lot::Mutex;
17use serde_json::{json, Map, Value};
18use std::path::{Path, PathBuf};
19use std::sync::Arc;
20use std::time::Duration;
22pub const FILE: &str = "keysync.json";
23pub const WRITE_LOCK: &str = "keysync.json.lock";
24pub const OWNER_LOCK: &str = "keysync.owner.lock";
25pub const INTERVAL: Duration = Duration::from_secs(90);
26/// Freshness of the verified server state (owner, 2026-11-05): own key pairs daily, imported public keys every 90 min.
27pub const FRESH_FULL_SECS: i64 = 24 * 3600;
28pub const FRESH_PUBLIC_SECS: i64 = 90 * 60;
29const MAX_JOBS_PER_RUN: usize = 4;
30const MAX_REFRESH_PER_RUN: usize = 3;
31const MAX_UNCONFIRMED: u64 = 5;
32const MAX_BACKOFF_SECS: i64 = 3600;
33const KEEP_FINISHED_SECS: i64 = 24 * 3600;
34/// A failed refresh is tried again after this pause instead of waiting for the full freshness period.
35const REFRESH_ERROR_PAUSE_SECS: i64 = 600;
36const MAX_JOBS: usize = 500;
38/// Server access. Production uses the strict HTTPS client of `keyserver.rs`; tests use a scripted server.
39pub enum Transport {
40 Https(reqwest::Client),
41 #[cfg(test)]
42 Fake(Arc<Mutex<tests::FakeServer>>),
45impl Transport {
46 pub fn https() -> CoreResult<Self> {
47 Ok(Self::Https(crate::core::keyserver::client()?))
48 }
49 async fn upload(&self, fingerprint: &str, keytext: &str) -> CoreResult<()> {
50 match self {
51 Self::Https(client) => {
52 let body = serde_json::to_vec(&json!({ "keytext": keytext }))?;
53 crate::core::keyserver::upload_with(client, fingerprint, body).await
54 }
55 #[cfg(test)]
56 Self::Fake(server) => server.lock().upload(fingerprint, keytext),
57 }
58 }
59 async fn lookup(&self, fingerprint: &str) -> CoreResult<Value> {
60 match self {
61 Self::Https(client) => crate::core::keyserver::lookup_with(client, fingerprint).await,
62 #[cfg(test)]
63 Self::Fake(server) => server.lock().lookup(fingerprint),
64 }
65 }
68pub struct KeySync {
69 path: PathBuf,
70 write_lock: PathBuf,
71 owner_path: PathBuf,
72 /// Held while this process may process jobs (see the module comment).
73 owner: Mutex<Option<std::fs::File>>,
74 state: Mutex<Value>,
75 pub notify: tokio::sync::Notify,
76 run_lock: tokio::sync::Mutex<()>,
79fn now() -> i64 {
80 chrono::Utc::now().timestamp()
83fn open_lock(path: &Path) -> std::io::Result<std::fs::File> {
84 let mut options = std::fs::OpenOptions::new();
85 options.create(true).read(true).write(true).truncate(false);
86 #[cfg(unix)]
87 {
88 use std::os::unix::fs::OpenOptionsExt;
89 options.mode(0o600);
90 }
91 options.open(path)
94fn valid(state: Value) -> Option<Value> {
95 Some(state).filter(|v| v["version"] == 1 && v["keys"].is_object() && v["jobs"].is_array())
98fn empty() -> Value {
99 json!({ "version": 1, "lastRunAt": null, "keys": {}, "jobs": [] })
102/// Number of subkeys in an armored certificate; the server must return at least as many.
103fn subkey_count(armored: &str) -> u64 {
104 use pgp::composed::{Deserializable, SignedPublicKey};
105 SignedPublicKey::from_armor_single(std::io::Cursor::new(armored.as_bytes()))
106 .map(|(key, _)| key.public_subkeys.len() as u64)
107 .unwrap_or(0)
110fn retryable(error: &CoreError) -> bool {
111 match error.code.as_str() {
112 "KEYSERVER_UNAVAILABLE" | "KEYSERVER_RATE_LIMITED" => true,
113 "KEYSERVER_REJECTED" => error.details.get("status").and_then(Value::as_u64).is_some_and(|s| s >= 500),
114 _ => false,
115 }
118impl KeySync {
119 pub fn new(data: &Path) -> Arc<Self> {
120 let path = data.join(FILE);
121 let state = crate::core::storage::read_json(&path).ok().and_then(valid).unwrap_or_else(empty);
122 let sync = Arc::new(Self { path, write_lock: data.join(WRITE_LOCK), owner_path: data.join(OWNER_LOCK), owner: Mutex::new(None),
123 state: Mutex::new(state), notify: tokio::sync::Notify::new(), run_lock: tokio::sync::Mutex::new(()) });
124 sync.ensure_owner();
125 sync
126 }
128 fn disk(&self) -> Option<Value> {
129 crate::core::storage::read_json(&self.path).ok().and_then(valid)
130 }
132 /// Take the processing role if the helper does not hold it right now. On taking it, the helper's
133 /// results on disk become the current state.
134 pub fn ensure_owner(&self) -> bool {
135 let mut owner = self.owner.lock();
136 if owner.is_some() {
137 return true;
138 }
139 // Without a lock file (read-only data folder) behave as the only process, as before.
140 let Ok(file) = open_lock(&self.owner_path) else { return true };
141 if file.try_lock().is_err() {
142 return false;
143 }
144 *owner = Some(file);
145 if let Some(current) = self.disk() {
146 *self.state.lock() = current;
147 }
148 true
149 }
151 fn save(&self, state: &Value) -> CoreResult<()> {
152 crate::core::storage::atomic_write_json(&self.path, state)
153 }
155 fn update<R>(&self, change: impl FnOnce(&mut Value) -> CoreResult<R>) -> CoreResult<R> {
156 let mut state = self.state.lock();
157 // Short exclusive section shared with the helper: read the newest file, change it, write it back.
158 let _guard = open_lock(&self.write_lock).and_then(|file| file.lock().map(|_| file)).ok();
159 let mut draft = self.disk().unwrap_or_else(|| state.clone());
160 let result = change(&mut draft)?;
161 self.save(&draft)?;
162 *state = draft;
163 Ok(result)
164 }
166 /// Mirror the public part of Keychain OpenPGP entries (raw vault entries). Keys removed from the Keychain
167 /// disappear unless a job for them is still open.
168 pub fn mirror(&self, entries: &[Value]) -> CoreResult<()> {
169 self.mirror_with(entries, true)
170 }
172 /// `prune` only for a complete Keychain snapshot; adding one key (enqueue) must not drop the others' records.
173 fn mirror_with(&self, entries: &[Value], prune: bool) -> CoreResult<()> {
174 let mut seen = std::collections::HashSet::new();
175 let mut mirrored = Map::new();
176 for entry in entries.iter().filter(|e| e["kind"] == "openpgp") {
177 let Ok((fingerprint, body)) = crate::core::keyserver::public_payload(entry) else { continue };
178 let keytext = serde_json::from_slice::<Value>(&body).ok().and_then(|v| v["keytext"].as_str().map(str::to_string)).unwrap_or_default();
179 seen.insert(fingerprint.clone());
180 mirrored.insert(fingerprint, json!({ "keychainId": entry["id"], "label": entry["label"], "hasPrivate": entry["private"].is_string(),
181 "revoked": entry["revoked"] == true, "public": keytext, "subkeys": subkey_count(entry["public"].as_str().unwrap_or("")) }));
182 }
183 let apply = |state: &mut Value| -> bool {
184 let mut changed = false;
185 let open: std::collections::HashSet<String> = state["jobs"].as_array().into_iter().flatten()
186 .filter(|j| !matches!(j["state"].as_str(), Some("success" | "failed"))).filter_map(|j| j["fingerprint"].as_str().map(str::to_string)).collect();
187 let keys = state["keys"].as_object_mut().expect("keys");
188 if prune {
189 keys.retain(|fingerprint, _| { let keep = seen.contains(fingerprint) || open.contains(fingerprint); changed |= !keep; keep });
190 }
191 for (fingerprint, fresh) in &mirrored {
192 let slot = keys.entry(fingerprint.clone()).or_insert_with(|| json!({}));
193 for field in ["keychainId", "label", "hasPrivate", "revoked", "public", "subkeys"] {
194 if slot[field] != fresh[field] { slot[field] = fresh[field].clone(); changed = true; }
195 }
196 }
197 changed
198 };
199 // Write only when the mirror differs; the write applies the same change to the newest file.
200 if apply(&mut self.state.lock().clone()) { self.update(|state| { apply(state); Ok(()) })?; }
201 Ok(())
202 }
204 /// Queue a publication, revocation or check for a raw Keychain entry. A revocation supersedes an open
205 /// publication of the same key; an identical open job is refreshed instead of duplicated.
206 pub fn enqueue(&self, kind: &str, entry: &Value) -> CoreResult<Value> {
207 if !["publish", "revoke", "check"].contains(&kind) {
208 return Err(CoreError::plain("Unknown key sync job."));
209 }
210 let (fingerprint, body) = crate::core::keyserver::public_payload(entry)?;
211 if kind == "revoke" && entry["revoked"] != true {
212 return Err(CoreError::new("KEYSYNC_NOT_REVOKED", "Revoke the key locally before publishing its revocation."));
213 }
214 let keytext = serde_json::from_slice::<Value>(&body)?["keytext"].as_str().unwrap_or("").to_string();
215 self.mirror_with(std::slice::from_ref(entry), false).ok();
216 let job = self.update(|state| {
217 let stamp = now();
218 let jobs = state["jobs"].as_array_mut().expect("jobs");
219 if kind == "revoke" {
220 jobs.retain(|j| !(j["fingerprint"] == fingerprint && j["kind"] == "publish" && !matches!(j["state"].as_str(), Some("success" | "failed"))));
221 }
222 if let Some(open) = jobs.iter_mut().find(|j| j["fingerprint"] == fingerprint && j["kind"] == kind && !matches!(j["state"].as_str(), Some("success" | "failed"))) {
223 // A running job keeps its payload; a waiting one takes the newest certificate and runs now.
224 if !matches!(open["state"].as_str(), Some("uploading" | "verifying")) {
225 open["keytext"] = json!(keytext); open["state"] = json!("pending"); open["nextAt"] = json!(stamp); open["updatedAt"] = json!(stamp);
226 }
227 return Ok(open.clone());
228 }
229 if jobs.len() >= MAX_JOBS { jobs.retain(|j| !matches!(j["state"].as_str(), Some("success"))); }
230 if jobs.len() >= MAX_JOBS {
231 return Err(CoreError::new("KEYSYNC_QUEUE_FULL", "The key server queue is full. Dismiss failed jobs first."));
232 }
233 let job = json!({ "id": uuid::Uuid::new_v4().to_string(), "fingerprint": fingerprint, "kind": kind, "state": "pending", "attempts": 0,
234 "unconfirmed": 0, "nextAt": stamp, "lastError": null, "createdAt": stamp, "updatedAt": stamp, "keytext": if kind == "check" { String::new() } else { keytext } });
235 jobs.push(job.clone());
236 Ok(job)
237 })?;
238 self.notify.notify_one();
239 Ok(public_job(&job))
240 }
242 pub fn retry(&self, job_id: &str) -> CoreResult<()> {
243 self.update(|state| {
244 let job = state["jobs"].as_array_mut().expect("jobs").iter_mut().find(|j| j["id"] == job_id && j["state"] == "failed")
245 .ok_or_else(|| CoreError::plain("Only failed jobs can be retried."))?;
246 job["state"] = json!("pending"); job["attempts"] = json!(0); job["unconfirmed"] = json!(0); job["nextAt"] = json!(now());
247 Ok(())
248 })?;
249 self.notify.notify_one();
250 Ok(())
251 }
253 pub fn dismiss(&self, job_id: &str) -> CoreResult<()> {
254 self.update(|state| {
255 state["jobs"].as_array_mut().expect("jobs").retain(|j| !(j["id"] == job_id && matches!(j["state"].as_str(), Some("success" | "failed"))));
256 Ok(())
257 })
258 }
260 /// Queue overview for the Keychain bar and per-key state (no certificate text).
261 pub fn summary(&self) -> Value {
262 let state = self.state.lock();
263 let jobs: Vec<Value> = state["jobs"].as_array().into_iter().flatten().map(public_job).collect();
264 let mut counts = json!({ "pending": 0, "uploading": 0, "verifying": 0, "success": 0, "retry": 0, "failed": 0 });
265 for job in &jobs { if let Some(key) = job["state"].as_str() { counts[key] = json!(counts[key].as_u64().unwrap_or(0) + 1); } }
266 let mut keys = Map::new();
267 for (fingerprint, key) in state["keys"].as_object().into_iter().flatten() {
268 let job = jobs.iter().rev().find(|j| j["fingerprint"] == fingerprint.as_str());
269 keys.insert(fingerprint.clone(), json!({ "server": key["server"], "job": job.map(|j| json!({ "kind": j["kind"], "state": j["state"], "lastError": j["lastError"] })) }));
270 }
271 json!({ "intervalSecs": INTERVAL.as_secs(), "lastRunAt": state["lastRunAt"], "counts": counts, "jobs": jobs, "keys": keys })
272 }
274 /// One pass: due jobs (revocations first), then stale verified states. Returns whether anything changed.
275 pub async fn run_once(&self, transport: &Transport) -> bool {
276 let _turn = self.run_lock.lock().await;
277 // The helper is finishing a pass right now; Core takes over at the next run.
278 if !self.ensure_owner() {
279 return false;
280 }
281 // Jobs left `uploading`/`verifying` by an interrupted pass (app or helper ended mid-job) would never be
282 // picked again: run_lock and the owner lock mean no pass is active now, so they run again (idempotent upload).
283 let interrupted = self.state.lock()["jobs"].as_array().into_iter().flatten().any(|j| matches!(j["state"].as_str(), Some("uploading" | "verifying")));
284 if interrupted {
285 let _ = self.update(|state| {
286 let stamp = now();
287 for job in state["jobs"].as_array_mut().expect("jobs").iter_mut().filter(|j| matches!(j["state"].as_str(), Some("uploading" | "verifying"))) {
288 job["state"] = json!("retry"); job["nextAt"] = json!(stamp); job["updatedAt"] = json!(stamp);
289 }
290 Ok(())
291 });
292 }
293 let stamp = now();
294 let due: Vec<Value> = {
295 let state = self.state.lock();
296 let mut due: Vec<Value> = state["jobs"].as_array().into_iter().flatten()
297 .filter(|j| matches!(j["state"].as_str(), Some("pending" | "retry")) && j["nextAt"].as_i64().unwrap_or(0) <= stamp).cloned().collect();
298 due.sort_by_key(|j| (j["kind"] != "revoke", j["createdAt"].as_i64().unwrap_or(0)));
299 due.truncate(MAX_JOBS_PER_RUN);
300 due
301 };
302 let mut changed = false;
303 for job in due {
304 changed = true;
305 self.process(transport, &job).await;
306 }
307 let stale: Vec<(String, bool)> = {
308 let state = self.state.lock();
309 let busy: std::collections::HashSet<&str> = state["jobs"].as_array().into_iter().flatten()
310 .filter(|j| !matches!(j["state"].as_str(), Some("success" | "failed"))).filter_map(|j| j["fingerprint"].as_str()).collect();
311 let mut stale: Vec<(String, bool, i64)> = state["keys"].as_object().into_iter().flatten().filter(|(fp, _)| !busy.contains(fp.as_str())).filter_map(|(fp, key)| {
312 let checked = key["server"]["checkedAt"].as_i64().unwrap_or(0);
313 let error_at = key["server"]["errorAt"].as_i64().unwrap_or(0);
314 let fresh = if key["hasPrivate"] == true { FRESH_FULL_SECS } else { FRESH_PUBLIC_SECS };
315 // The helper compares packets without rPGP; Core confirms its states with a full check.
316 let by_helper = key["server"]["by"] == "helper";
317 ((by_helper || stamp - checked >= fresh) && stamp - error_at >= REFRESH_ERROR_PAUSE_SECS).then(|| (fp.clone(), key["revoked"] == true, checked))
318 }).collect();
319 stale.sort_by_key(|(_, _, checked)| *checked);
320 stale.into_iter().take(MAX_REFRESH_PER_RUN).map(|(fp, revoked, _)| (fp, revoked)).collect()
321 };
322 for (fingerprint, _) in stale {
323 changed = true;
324 match transport.lookup(&fingerprint).await {
325 Ok(remote) => { let _ = self.record_server(&fingerprint, &remote); }
326 // The last verified state stays; only the error and its time are noted.
327 Err(error) => { let _ = self.update(|state| { if let Some(key) = state["keys"].get_mut(&fingerprint) { if !key["server"].is_object() { key["server"] = json!({}); } key["server"]["errorAt"] = json!(now()); key["server"]["error"] = json!(error.code); } Ok(()) }); }
328 }
329 }
330 let _ = self.update(|state| {
331 state["lastRunAt"] = json!(now());
332 let limit = now() - KEEP_FINISHED_SECS;
333 state["jobs"].as_array_mut().expect("jobs").retain(|j| !(j["state"] == "success" && j["updatedAt"].as_i64().unwrap_or(0) < limit));
334 Ok(())
335 });
336 changed
337 }
339 fn set_job(&self, id: &str, change: impl FnOnce(&mut Value)) {
340 let _ = self.update(|state| {
341 if let Some(job) = state["jobs"].as_array_mut().expect("jobs").iter_mut().find(|j| j["id"] == id) { change(job); job["updatedAt"] = json!(now()); }
342 Ok(())
343 });
344 }
346 fn record_server(&self, fingerprint: &str, remote: &Value) -> CoreResult<()> {
347 self.update(|state| {
348 if let Some(key) = state["keys"].get_mut(fingerprint) {
349 let stamp = now();
350 key["server"] = json!({ "checkedAt": stamp, "verifiedAt": stamp, "found": remote["found"], "revoked": remote["revoked"], "subkeys": remote["subkeys"] });
351 }
352 Ok(())
353 })
354 }
356 async fn process(&self, transport: &Transport, job: &Value) {
357 let id = job["id"].as_str().unwrap_or("").to_string();
358 let fingerprint = job["fingerprint"].as_str().unwrap_or("").to_string();
359 let kind = job["kind"].as_str().unwrap_or("").to_string();
360 let expected_subkeys = self.state.lock()["keys"][&fingerprint]["subkeys"].as_u64().unwrap_or(0);
361 let outcome: CoreResult<Value> = async {
362 if kind != "check" {
363 self.set_job(&id, |j| j["state"] = json!("uploading"));
364 transport.upload(&fingerprint, job["keytext"].as_str().unwrap_or("")).await?;
365 }
366 self.set_job(&id, |j| j["state"] = json!("verifying"));
367 transport.lookup(&fingerprint).await
368 }.await;
369 match outcome {
370 Ok(remote) => {
371 let _ = self.record_server(&fingerprint, &remote);
372 let confirmed = match kind.as_str() {
373 "check" => true,
374 "revoke" => remote["found"] == true && remote["revoked"] == true,
375 // A server that strips subkeys or signatures (seen with PQ certificates) is not a success.
376 _ => remote["found"] == true && remote["subkeys"].as_u64().unwrap_or(0) >= expected_subkeys,
377 };
378 self.set_job(&id, |j| {
379 if confirmed { j["state"] = json!("success"); j["lastError"] = Value::Null; return; }
380 let unconfirmed = j["unconfirmed"].as_u64().unwrap_or(0) + 1;
381 j["unconfirmed"] = json!(unconfirmed);
382 j["lastError"] = json!("KEYSERVER_NOT_CONFIRMED");
383 if unconfirmed >= MAX_UNCONFIRMED { j["state"] = json!("failed"); } else { j["state"] = json!("retry"); j["nextAt"] = json!(now() + INTERVAL.as_secs() as i64); }
384 });
385 }
386 Err(error) => {
387 let again = retryable(&error);
388 self.set_job(&id, |j| {
389 let attempts = j["attempts"].as_u64().unwrap_or(0) + 1;
390 j["attempts"] = json!(attempts);
391 j["lastError"] = json!(error.code);
392 if again {
393 let backoff = (INTERVAL.as_secs() as i64).saturating_mul(1 << attempts.min(6)).min(MAX_BACKOFF_SECS);
394 j["state"] = json!("retry"); j["nextAt"] = json!(now() + backoff);
395 } else { j["state"] = json!("failed"); }
396 });
397 }
398 }
399 }
401 /// Server state for a quick verification (owner 2026-11-05): a stored state younger than the freshness
402 /// period (24 h key pair, 90 min public key) is shown as is; an older one is checked on the server now.
403 pub async fn status_for_verify(&self, transport: &Transport, fingerprint: &str) -> Value {
404 let fingerprint = fingerprint.to_ascii_uppercase();
405 let (server, period) = {
406 let state = self.state.lock();
407 let Some(key) = state["keys"].get(&fingerprint) else { return json!({ "known": false }) };
408 (key["server"].clone(), if key["hasPrivate"] == true { FRESH_FULL_SECS } else { FRESH_PUBLIC_SECS })
409 };
410 let verified = server["verifiedAt"].as_i64();
411 // A helper result is only a packet comparison; Core shows it as fresh after its own verification.
412 if server["by"] != "helper" && verified.is_some_and(|at| now() - at < period) {
413 return json!({ "known": true, "found": server["found"], "revoked": server["revoked"], "verifiedAt": verified, "fresh": true, "live": false });
414 }
415 match tokio::time::timeout(Duration::from_secs(12), transport.lookup(&fingerprint)).await {
416 Ok(Ok(remote)) => {
417 let _ = self.record_server(&fingerprint, &remote);
418 json!({ "known": true, "found": remote["found"], "revoked": remote["revoked"], "verifiedAt": now(), "fresh": true, "live": true })
419 }
420 outcome => {
421 let code = match outcome { Ok(Err(error)) => error.code, _ => "KEYSERVER_UNAVAILABLE".to_string() };
422 json!({ "known": true, "found": server["found"], "revoked": server["revoked"], "verifiedAt": verified, "fresh": false, "live": false, "error": code })
423 }
424 }
425 }
427 /// Keyserver state for a verification with any key: Keychain identities use `status_for_verify`; other keys
428 /// (saved in Hash, picked file, downloaded now) are looked up live, without recording anything.
429 pub async fn server_status_any(&self, transport: &Transport, fingerprint: &str) -> Value {
430 let tracked = self.status_for_verify(transport, fingerprint).await;
431 if tracked["known"] == true {
432 return tracked;
433 }
434 match tokio::time::timeout(Duration::from_secs(12), transport.lookup(&fingerprint.to_ascii_uppercase())).await {
435 Ok(Ok(remote)) => json!({ "known": true, "tracked": false, "found": remote["found"], "revoked": remote["revoked"], "verifiedAt": now(), "fresh": true, "live": true }),
436 outcome => {
437 let code = match outcome { Ok(Err(error)) => error.code, _ => "KEYSERVER_UNAVAILABLE".to_string() };
438 json!({ "known": true, "tracked": false, "found": null, "revoked": null, "fresh": false, "live": false, "error": code })
439 }
440 }
441 }
443 /// Copy verified server states into the vault's per-key `keyserver` record (only while unlocked and only
444 /// when the stored certificate is still the one that was checked).
445 pub fn apply_to_vault(&self, vault: &crate::core::vault::VaultStore) {
446 let Ok(generation) = vault.unlocked_generation() else { return };
447 let keys = self.state.lock()["keys"].clone();
448 let _ = vault.update_at_generation(generation, |data| {
449 let Some(items) = data.pointer_mut("/secrets/keychain").and_then(Value::as_array_mut) else { return Ok(()) };
450 for entry in items.iter_mut().filter(|e| e["kind"] == "openpgp") {
451 let Some(fingerprint) = entry["fingerprint"].as_str() else { continue };
452 // Helper results wait for Core's own verification (see run_once) before they reach the vault.
453 let Some(server) = keys.get(fingerprint).map(|k| &k["server"]).filter(|s| s["verifiedAt"].is_i64() && s["by"] != "helper") else { continue };
454 let checked = chrono::DateTime::from_timestamp(server["verifiedAt"].as_i64().unwrap_or(0), 0).map(|d| d.to_rfc3339()).unwrap_or_default();
455 if entry["keyserver"]["checkedAt"].as_str().is_some_and(|c| c >= checked.as_str()) { continue; }
456 if !entry["keyserver"].is_object() { entry["keyserver"] = json!({}); }
457 entry["keyserver"]["origin"] = json!(crate::core::keyserver::ORIGIN);
458 entry["keyserver"]["checkedAt"] = json!(checked);
459 entry["keyserver"]["found"] = server["found"].clone();
460 entry["keyserver"]["revoked"] = server["revoked"].clone();
461 if server["found"] == true { entry["keyserver"][if server["revoked"] == true { "revocationPublishedAt" } else { "publishedAt" }] = json!(checked); }
462 }
463 Ok(())
464 });
465 }
468fn public_job(job: &Value) -> Value {
469 json!({ "id": job["id"], "fingerprint": job["fingerprint"], "kind": job["kind"], "state": job["state"], "attempts": job["attempts"],
470 "nextAt": job["nextAt"], "lastError": job["lastError"], "createdAt": job["createdAt"], "updatedAt": job["updatedAt"] })
473#[cfg(test)]
474pub mod tests {
475 use super::*;
477 /// Scripted keyserver: stores uploads, can fail, can strip subkeys like the PQ incident.
478 #[derive(Default)]
479 pub struct FakeServer {
480 pub stored: std::collections::HashMap<String, (bool, u64)>,
481 pub fail_with: Option<(String, Option<u16>)>,
482 pub strip_subkeys: bool,
483 pub uploads: usize,
484 }
485 impl FakeServer {
486 pub fn upload(&mut self, fingerprint: &str, keytext: &str) -> CoreResult<()> {
487 if let Some((code, status)) = self.fail_with.clone() {
488 let error = CoreError::new(&code, "scripted failure");
489 return Err(match status { Some(s) => error.with("status", s), None => error });
490 }
491 self.uploads += 1;
492 use pgp::composed::{Deserializable, SignedPublicKey};
493 let (key, _) = SignedPublicKey::from_armor_single(std::io::Cursor::new(keytext.as_bytes())).map_err(|_| CoreError::plain("bad"))?;
494 let revoked = !key.details.revocation_signatures.is_empty() || self.stored.get(fingerprint).is_some_and(|s| s.0);
495 self.stored.insert(fingerprint.to_string(), (revoked, if self.strip_subkeys { 0 } else { key.public_subkeys.len() as u64 }));
496 Ok(())
497 }
498 pub fn lookup(&mut self, fingerprint: &str) -> CoreResult<Value> {
499 if let Some((code, _)) = self.fail_with.clone() { return Err(CoreError::new(&code, "scripted failure")); }
500 Ok(match self.stored.get(fingerprint) { Some((revoked, subkeys)) => json!({"found":true,"revoked":revoked,"subkeys":subkeys}), None => json!({"found":false,"revoked":false,"subkeys":0}) })
501 }
502 }
504 fn entry() -> Value {
505 let (_, entry) = crate::core::openpgp::generate(&json!({"name":"KeySync fixture","subkeys":[{"role":"encrypt","algorithm":"curve25519"}]})).unwrap();
506 entry
507 }
508 fn fake() -> (Arc<Mutex<FakeServer>>, Transport) { let server = Arc::new(Mutex::new(FakeServer::default())); (server.clone(), Transport::Fake(server)) }
509 fn job_state(sync: &KeySync, kind: &str) -> String {
510 sync.summary()["jobs"].as_array().unwrap().iter().rev().find(|j| j["kind"] == kind).map(|j| j["state"].as_str().unwrap().to_string()).unwrap_or_default()
511 }
512 fn rewind(sync: &KeySync) { let _ = sync.update(|s| { for j in s["jobs"].as_array_mut().unwrap() { j["nextAt"] = json!(0); } Ok(()) }); }
514 #[tokio::test]
515 async fn interrupted_jobs_run_again_and_enqueue_keeps_other_keys() {
516 let dir = tempfile::tempdir().unwrap();
517 let sync = KeySync::new(dir.path());
518 let (a, b) = (entry(), entry());
519 sync.mirror(&[a.clone(), b.clone()]).unwrap();
520 let fp_b = b["fingerprint"].as_str().unwrap().to_string();
521 let _ = sync.update(|s| { s["keys"][&fp_b]["server"] = json!({"verifiedAt": now(), "found": true, "revoked": false}); Ok(()) });
522 sync.enqueue("publish", &a).unwrap();
523 assert_eq!(sync.summary()["keys"][&fp_b]["server"]["found"], true, "queueing A must not drop B's record");
524 // A pass ended mid-upload (app quit, helper stopped for an update).
525 let _ = sync.update(|s| { s["jobs"][0]["state"] = json!("uploading"); s["jobs"][0]["nextAt"] = json!(now() + 3600); Ok(()) });
526 let (server, transport) = fake();
527 assert!(sync.run_once(&transport).await);
528 assert_eq!(job_state(&sync, "publish"), "success");
529 assert_eq!(server.lock().uploads, 1);
530 }
532 #[tokio::test]
533 async fn helper_results_are_not_shown_as_verified_before_core_checks_them() {
534 let dir = tempfile::tempdir().unwrap();
535 let sync = KeySync::new(dir.path());
536 let a = entry();
537 let fp = a["fingerprint"].as_str().unwrap().to_string();
538 sync.mirror(std::slice::from_ref(&a)).unwrap();
539 let _ = sync.update(|s| { s["keys"][&fp]["server"] = json!({"verifiedAt": now(), "found": true, "revoked": true, "by": "helper"}); Ok(()) });
540 let (_server, transport) = fake();
541 let status = sync.status_for_verify(&transport, &fp).await;
542 assert_eq!((status["live"].as_bool(), status["revoked"].as_bool()), (Some(true), Some(false)), "{status}");
543 }
545 #[tokio::test]
546 async fn publish_succeeds_only_after_the_certificate_is_fetched_back_and_persists() {
547 let dir = tempfile::tempdir().unwrap();
548 let sync = KeySync::new(dir.path());
549 let entry = entry();
550 let (server, transport) = fake();
551 sync.enqueue("publish", &entry).unwrap();
552 assert_eq!(job_state(&sync, "publish"), "pending");
553 assert!(sync.run_once(&transport).await);
554 assert_eq!(job_state(&sync, "publish"), "success");
555 assert_eq!(server.lock().uploads, 1);
556 let fp = entry["fingerprint"].as_str().unwrap();
557 assert_eq!(sync.summary()["keys"][fp]["server"]["found"], true);
558 // Persisted public-only file, readable again (e.g. by the helper).
559 let text = std::fs::read_to_string(dir.path().join(FILE)).unwrap();
560 assert!(text.contains("PGP PUBLIC KEY BLOCK") && !text.contains("PRIVATE KEY"));
561 assert_eq!(KeySync::new(dir.path()).summary()["counts"]["success"], 1);
562 }
564 #[tokio::test]
565 async fn stripped_certificate_is_never_success_and_network_errors_back_off() {
566 let dir = tempfile::tempdir().unwrap();
567 let sync = KeySync::new(dir.path());
568 let entry = entry();
569 let (server, transport) = fake();
570 server.lock().strip_subkeys = true;
571 sync.enqueue("publish", &entry).unwrap();
572 for _ in 0..MAX_UNCONFIRMED { rewind(&sync); sync.run_once(&transport).await; }
573 assert_eq!(job_state(&sync, "publish"), "failed");
574 server.lock().strip_subkeys = false;
575 server.lock().fail_with = Some(("KEYSERVER_UNAVAILABLE".into(), None));
576 sync.enqueue("check", &entry).unwrap();
577 sync.run_once(&transport).await;
578 assert_eq!(job_state(&sync, "check"), "retry");
579 let next = sync.summary()["jobs"].as_array().unwrap().iter().find(|j| j["kind"] == "check").unwrap()["nextAt"].as_i64().unwrap();
580 assert!(next >= now() + 90);
581 server.lock().fail_with = Some(("KEYSERVER_UNSUPPORTED_FORMAT".into(), Some(415)));
582 rewind(&sync); sync.run_once(&transport).await;
583 assert_eq!(job_state(&sync, "check"), "failed");
584 }
586 #[tokio::test]
587 async fn running_job_is_not_duplicated_and_failed_refresh_retries_after_a_pause() {
588 let dir = tempfile::tempdir().unwrap();
589 let sync = KeySync::new(dir.path());
590 let entry = entry();
591 sync.enqueue("check", &entry).unwrap();
592 let _ = sync.update(|s| { s["jobs"][0]["state"] = json!("verifying"); Ok(()) });
593 sync.enqueue("check", &entry).unwrap();
594 assert_eq!(sync.summary()["jobs"].as_array().unwrap().len(), 1);
595 let _ = sync.update(|s| { s["jobs"] = json!([]); Ok(()) });
596 let (server, transport) = fake();
597 server.lock().fail_with = Some(("KEYSERVER_UNAVAILABLE".into(), None));
598 sync.run_once(&transport).await;
599 let fp = entry["fingerprint"].as_str().unwrap();
600 assert_eq!(sync.summary()["keys"][fp]["server"]["checkedAt"], Value::Null);
601 server.lock().fail_with = None;
602 sync.run_once(&transport).await;
603 assert_eq!(sync.summary()["keys"][fp]["server"]["found"], Value::Null, "paused after an error");
604 let _ = sync.update(|s| { s["keys"][fp]["server"]["errorAt"] = json!(now() - REFRESH_ERROR_PAUSE_SECS); Ok(()) });
605 sync.run_once(&transport).await;
606 assert_eq!(sync.summary()["keys"][fp]["server"]["found"], false);
607 }
609 #[tokio::test]
610 async fn revocation_supersedes_publication_and_needs_a_revoked_server_copy() {
611 let dir = tempfile::tempdir().unwrap();
612 let sync = KeySync::new(dir.path());
613 let entry = entry();
614 let (server, transport) = fake();
615 assert_eq!(sync.enqueue("revoke", &entry).unwrap_err().code, "KEYSYNC_NOT_REVOKED");
616 sync.enqueue("publish", &entry).unwrap();
617 let revoked = crate::core::keychain::revoke(&entry, "").unwrap();
618 sync.enqueue("revoke", &revoked).unwrap();
619 let jobs = sync.summary()["jobs"].as_array().unwrap().clone();
620 assert_eq!(jobs.len(), 1);
621 assert_eq!(jobs[0]["kind"], "revoke");
622 sync.run_once(&transport).await;
623 assert_eq!(job_state(&sync, "revoke"), "success");
624 assert_eq!(server.lock().stored[revoked["fingerprint"].as_str().unwrap()].0, true);
625 }
627 #[tokio::test]
628 async fn owner_lock_keeps_core_from_processing_while_the_helper_runs_and_writes_merge() {
629 let dir = tempfile::tempdir().unwrap();
630 // The helper holds the processing role before the app starts.
631 let helper = open_lock(&dir.path().join(OWNER_LOCK)).unwrap();
632 helper.try_lock().unwrap();
633 let sync = KeySync::new(dir.path());
634 let entry = entry();
635 let (server, transport) = fake();
636 sync.enqueue("publish", &entry).unwrap();
637 assert!(!sync.run_once(&transport).await);
638 assert_eq!(server.lock().uploads, 0);
639 // The helper finishes the job in the file; Core's later write must keep that result.
640 let mut disk: Value = serde_json::from_str(&std::fs::read_to_string(dir.path().join(FILE)).unwrap()).unwrap();
641 disk["jobs"][0]["state"] = json!("success");
642 std::fs::write(dir.path().join(FILE), disk.to_string()).unwrap();
643 let mut other = self::entry(); other["private"] = Value::Null;
644 sync.mirror(&[entry.clone(), other]).unwrap();
645 drop(helper);
646 assert!(sync.ensure_owner());
647 assert_eq!(job_state(&sync, "publish"), "success");
648 assert_eq!(sync.summary()["keys"].as_object().unwrap().len(), 2);
649 }
651 #[tokio::test]
652 async fn verification_uses_a_fresh_state_and_checks_a_stale_one_on_the_server() {
653 let dir = tempfile::tempdir().unwrap();
654 let sync = KeySync::new(dir.path());
655 let mut public_only = entry(); public_only["private"] = Value::Null;
656 let fp = public_only["fingerprint"].as_str().unwrap().to_string();
657 sync.mirror(std::slice::from_ref(&public_only)).unwrap();
658 let (server, transport) = fake();
659 assert_eq!(sync.status_for_verify(&transport, "0".repeat(40).as_str()).await["known"], false);
660 server.lock().stored.insert(fp.clone(), (true, 1));
661 let first = sync.status_for_verify(&transport, &fp).await;
662 assert_eq!((first["live"].clone(), first["revoked"].clone()), (json!(true), json!(true)));
663 server.lock().stored.clear();
664 let cached = sync.status_for_verify(&transport, &fp).await;
665 assert_eq!((cached["live"].clone(), cached["revoked"].clone()), (json!(false), json!(true)));
666 // Older than 90 min for a public-only key: checked again; a network error keeps the old state, marked stale.
667 let _ = sync.update(|s| { s["keys"][&fp]["server"]["verifiedAt"] = json!(now() - FRESH_PUBLIC_SECS); Ok(()) });
668 server.lock().fail_with = Some(("KEYSERVER_UNAVAILABLE".into(), None));
669 let stale = sync.status_for_verify(&transport, &fp).await;
670 assert_eq!((stale["fresh"].clone(), stale["error"].clone(), stale["revoked"].clone()), (json!(false), json!("KEYSERVER_UNAVAILABLE"), json!(true)));
671 // Helper-recorded states are refreshed by Core at its next run.
672 let _ = sync.update(|s| { s["keys"][&fp]["server"] = json!({"checkedAt": now(), "verifiedAt": now(), "found": true, "revoked": true, "by": "helper"}); Ok(()) });
673 server.lock().fail_with = None;
674 sync.run_once(&transport).await;
675 assert_eq!(sync.summary()["keys"][&fp]["server"]["found"], false);
676 }
678 #[tokio::test]
679 async fn stale_states_refresh_by_freshness_and_the_vault_gets_the_verified_state() {
680 let dir = tempfile::tempdir().unwrap();
681 let sync = KeySync::new(dir.path());
682 let full = entry();
683 let mut public_only = entry(); public_only["private"] = Value::Null;
684 sync.mirror(&[full.clone(), public_only.clone()]).unwrap();
685 let (server, transport) = fake();
686 server.lock().stored.insert(public_only["fingerprint"].as_str().unwrap().into(), (false, 1));
687 sync.run_once(&transport).await;
688 let fp_full = full["fingerprint"].as_str().unwrap(); let fp_pub = public_only["fingerprint"].as_str().unwrap();
689 assert_eq!(sync.summary()["keys"][fp_pub]["server"]["found"], true);
690 // Checked 2 h ago: the public-only key (90 min) is stale, the key pair (24 h) is not.
691 let _ = sync.update(|s| { for fp in [fp_full, fp_pub] { s["keys"][fp]["server"]["checkedAt"] = json!(now() - 7200); } Ok(()) });
692 server.lock().stored.clear();
693 sync.run_once(&transport).await;
694 assert_eq!(sync.summary()["keys"][fp_pub]["server"]["found"], false);
695 assert_eq!(sync.summary()["keys"][fp_full]["server"]["checkedAt"].as_i64().unwrap() <= now() - 7000, true);
696 // Removed from the Keychain without an open job: dropped from the mirror.
697 sync.mirror(&[full.clone()]).unwrap();
698 assert!(sync.summary()["keys"].get(fp_pub).is_none());
699 // Verified state reaches the vault record.
700 let vault = crate::core::vault::VaultStore::new(dir.path().join("vault-root").as_path());
701 vault.create("fixture-vault-password").await.unwrap();
702 crate::core::keychain::store(&vault, full.clone()).unwrap();
703 sync.apply_to_vault(&vault);
704 assert_eq!(crate::core::keychain::list(&vault).unwrap()[0]["keyserver"]["found"], false);
705 }
SHA-2568ae86182763a6c577f076296e40692ad1c65a2820dd16bab58c8726f19e6ad47

CH-J Proprietary Software License 1.14 · 源代码依据 CH-J Proprietary Software License 1.14 提供。公开源代码不会改变许可条款，也不会授予额外权利。

压缩包 SHA-256ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f
文件