From f09300fd9529aecc8d49090aac0303bf5774ac28 Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Mon, 12 Oct 2026 09:07:32 +0100 Subject: [PATCH] fix: a panic on one thread no longer takes the whole app down Every shared lock was std::sync::Mutex with lock().unwrap() (~150 sites), so a panic in any session's reader or callback poisoned the lock and the next UI frame panicked too, dropping every open session. Locks are now parking_lot::Mutex (no poisoning; already in the dependency tree), and a panic hook reports worker panics as an error toast instead of the session silently stopping (F3, F4). --- rust/Cargo.lock | 2 + rust/app/Cargo.toml | 1 + rust/app/src/app.rs | 16 ++++-- rust/app/src/chrome.rs | 6 +-- rust/app/src/forwards.rs | 9 ++-- rust/app/src/github_view.rs | 9 ++-- rust/app/src/main.rs | 10 ++++ rust/app/src/profile_editor.rs | 7 +-- rust/app/src/settings.rs | 7 +-- rust/app/src/sftp.rs | 9 ++-- rust/app/src/terminal.rs | 9 ++-- rust/app/src/vnc_view.rs | 2 +- rust/core/Cargo.toml | 1 + rust/core/src/clients/github.rs | 15 +++--- rust/core/src/clients/irc.rs | 35 +++++++------ rust/core/src/clients/local.rs | 30 +++++------ rust/core/src/clients/mod.rs | 12 ++--- rust/core/src/clients/raw.rs | 9 ++-- rust/core/src/clients/serial.rs | 13 ++--- rust/core/src/clients/ssh.rs | 53 +++++++++---------- rust/core/src/clients/telnet.rs | 17 +++--- rust/core/src/clients/vnc.rs | 21 ++++---- rust/core/src/recording.rs | 8 +-- rust/core/src/secrets.rs | 18 +++---- rust/core/src/sessions.rs | 83 +++++++++++++++--------------- rust/core/src/storage.rs | 43 +++++++--------- rust/core/tests/sessions.rs | 5 +- rust/core/tests/ssh_integration.rs | 23 +++++---- 28 files changed, 252 insertions(+), 221 deletions(-) diff --git a/rust/Cargo.lock b/rust/Cargo.lock index be8aab0..5ea4c5d 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -2806,6 +2806,7 @@ dependencies = [ "egui-phosphor", "image", "nexus-core", + "parking_lot", "regex", "rfd", "serde", @@ -2827,6 +2828,7 @@ dependencies = [ "image", "keyring", "libc", + "parking_lot", "portable-pty", "rand 0.8.8", "rusqlite", diff --git a/rust/app/Cargo.toml b/rust/app/Cargo.toml index e1fe582..1c7bd7e 100644 --- a/rust/app/Cargo.toml +++ b/rust/app/Cargo.toml @@ -12,6 +12,7 @@ path = "src/main.rs" nexus-core = { path = "../core" } eframe = { version = "0.36", default-features = false, features = ["default_fonts", "glow", "wayland", "x11", "accesskit", "persistence"] } egui = "0.36" +parking_lot = "0.12" egui-phosphor = "0.14" rfd = "0.17" image = { version = "0.25", default-features = false, features = ["png"] } diff --git a/rust/app/src/app.rs b/rust/app/src/app.rs index 478e19b..0ce234b 100644 --- a/rust/app/src/app.rs +++ b/rust/app/src/app.rs @@ -11,9 +11,10 @@ use egui::{Key, KeyboardShortcut, Modifiers, Ui}; use nexus_core::models::{Profile, Protocol, SessionInfo, SessionState}; use nexus_core::service::Service; use nexus_core::sessions::{Event, HostKeyDecision}; +use parking_lot::Mutex; use std::collections::VecDeque; use std::path::PathBuf; -use std::sync::{Arc, Mutex, mpsc}; +use std::sync::{Arc, mpsc}; pub struct Options { pub screenshot: Option, @@ -37,7 +38,14 @@ pub struct Options { static POSTED: Mutex> = Mutex::new(Vec::new()); pub fn post_error(e: String) { - POSTED.lock().unwrap().push(e); + POSTED.lock().push(e); +} + +/// For the panic hook: the panicking thread may already hold the lock. +pub fn post_error_nonblocking(e: String) { + if let Some(mut p) = POSTED.try_lock() { + p.push(e); + } } /// A credential or host-key question from a connecting session. @@ -141,7 +149,7 @@ impl App { let ctx = cc.egui_ctx.clone(); let tx = Mutex::new(tx); svc.sessions.set_sink(Arc::new(move |e| { - let _ = tx.lock().unwrap().send(e); + let _ = tx.lock().send(e); ctx.request_repaint(); })); let s = Shared::new(svc.clone(), cc.egui_ctx.clone()); @@ -405,7 +413,7 @@ impl App { // --- events --------------------------------------------------------------------------- fn drain_events(&mut self) { - for e in POSTED.lock().unwrap().drain(..) { + for e in POSTED.lock().drain(..) { self.s.error(e); } while let Ok(e) = self.rx.try_recv() { diff --git a/rust/app/src/chrome.rs b/rust/app/src/chrome.rs index 8d217c5..77c8494 100644 --- a/rust/app/src/chrome.rs +++ b/rust/app/src/chrome.rs @@ -370,13 +370,13 @@ pub fn status_bar(app: &mut App, ui: &mut Ui) { }); } -static LATENCY: std::sync::Mutex)>> = std::sync::Mutex::new(None); +static LATENCY: parking_lot::Mutex)>> = parking_lot::Mutex::new(None); /// Pings the active SSH session every 10 s for the status-bar latency. fn latency_tick(app: &mut App) { let id = app.s.active.clone(); let ssh = id.as_deref().and_then(|a| app.s.pane(a)).is_some_and(|p| p.info.protocol == Protocol::Ssh); - if let Some(r) = LATENCY.lock().unwrap().take() { + if let Some(r) = LATENCY.lock().take() { app.latency = Some(r); } let Some(id) = id.filter(|_| ssh) else { return }; @@ -395,7 +395,7 @@ fn latency_tick(app: &mut App) { let (svc, ctx) = (app.s.svc.clone(), app.s.ctx.clone()); std::thread::spawn(move || { let r = svc.sessions.ping(&id).map_err(|e| e.to_string()); - *LATENCY.lock().unwrap() = Some((id, r)); + *LATENCY.lock() = Some((id, r)); ctx.request_repaint(); }); } diff --git a/rust/app/src/forwards.rs b/rust/app/src/forwards.rs index 0554554..246ada6 100644 --- a/rust/app/src/forwards.rs +++ b/rust/app/src/forwards.rs @@ -7,7 +7,8 @@ use crate::widgets; use egui::{Align, Key, Layout, RichText}; use egui_phosphor::regular as icon; use nexus_core::models::{ForwardDirection, PortForward, new_id}; -use std::sync::{Arc, Mutex}; +use parking_lot::Mutex; +use std::sync::Arc; struct Form { remote: bool, @@ -49,7 +50,7 @@ impl Forwards { } pub fn ui(&mut self, ctx: &egui::Context, s: &mut Shared) { - for started in std::mem::take(&mut *self.done.lock().unwrap()) { + for started in std::mem::take(&mut *self.done.lock()) { if started { s.success("Forward started"); self.form.bind_port.clear(); @@ -131,7 +132,7 @@ impl Forwards { if let Some(id) = stop { let (svc, sid, done) = (s.svc.clone(), sid.clone(), self.done.clone()); - s.spawn(move || svc.sessions.stop_forward(&sid, &id).map_err(|e| format!("Failed to stop: {e}").into()), move |_| done.lock().unwrap().push(false)); + s.spawn(move || svc.sessions.stop_forward(&sid, &id).map_err(|e| format!("Failed to stop: {e}").into()), move |_| done.lock().push(false)); } if start { let f = &self.form; @@ -153,7 +154,7 @@ impl Forwards { let (svc, done) = (s.svc.clone(), self.done.clone()); s.spawn( move || svc.sessions.start_forward(&sid, rule).map_err(|e| format!("Failed to start forward: {e}").into()), - move |_| done.lock().unwrap().push(true), + move |_| done.lock().push(true), ); } _ => s.error("Bind port and destination port are required"), diff --git a/rust/app/src/github_view.rs b/rust/app/src/github_view.rs index c14e843..64c7027 100644 --- a/rust/app/src/github_view.rs +++ b/rust/app/src/github_view.rs @@ -8,7 +8,8 @@ use chrono::{DateTime, Utc}; use egui::{Color32, RichText, Sense, Ui, vec2}; use egui_phosphor::regular as icon; use nexus_core::models::GitHubRepo; -use std::sync::{Arc, Mutex}; +use parking_lot::Mutex; +use std::sync::Arc; enum Msg { Repos(Vec), @@ -175,7 +176,7 @@ impl GithubView { Ok(r) => Msg::Repos(r), Err(e) => Msg::RefreshFailed(e.to_string()), }; - inbox.lock().unwrap().push(msg); + inbox.lock().push(msg); ctx.request_repaint(); }); } @@ -193,13 +194,13 @@ impl GithubView { Ok(()) => Msg::Downloaded(full, path.display().to_string()), Err(e) => Msg::DownloadFailed(e.to_string()), }; - inbox.lock().unwrap().push(msg); + inbox.lock().push(msg); ctx.request_repaint(); }); } pub fn ui(&mut self, ui: &mut Ui, s: &mut Shared) { - let msgs = std::mem::take(&mut *self.inbox.lock().unwrap()); + let msgs = std::mem::take(&mut *self.inbox.lock()); for m in msgs { match m { Msg::Repos(r) => { diff --git a/rust/app/src/main.rs b/rust/app/src/main.rs index d6faa54..7240a6e 100644 --- a/rust/app/src/main.rs +++ b/rust/app/src/main.rs @@ -33,6 +33,16 @@ use nexus_core::service::Service; use std::sync::Arc; fn main() -> eframe::Result { + // A panic on a worker thread (a session's reader, a transfer) ends only that thread; say so in + // the UI rather than letting the session just stop. + let default_hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |info| { + default_hook(info); + let what = + info.payload().downcast_ref::<&str>().copied().or_else(|| info.payload().downcast_ref::().map(String::as_str)).unwrap_or("unknown error"); + let thread = std::thread::current().name().unwrap_or("worker").to_string(); + app::post_error_nonblocking(format!("Internal error in {thread} thread: {what}. Other sessions are unaffected; reconnect the affected one.")); + })); let mut opts = app::Options { screenshot: None, theme: None, diff --git a/rust/app/src/profile_editor.rs b/rust/app/src/profile_editor.rs index c9025a5..6be2b2e 100644 --- a/rust/app/src/profile_editor.rs +++ b/rust/app/src/profile_editor.rs @@ -11,9 +11,10 @@ use nexus_core::clients::serial; use nexus_core::keyfiles; use nexus_core::models::*; use nexus_core::validation::{ValidationError, validate_profile}; +use parking_lot::Mutex; use std::collections::{BTreeMap, HashMap}; use std::path::Path; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; type KeyChecks = HashMap>; @@ -184,7 +185,7 @@ impl ProfileEditor { /// Drawn inside the right dock while open. pub fn ui(&mut self, ui: &mut Ui, s: &mut Shared) { let t = Tokens::get(ui.ctx()); - if let Some(r) = self.test.lock().unwrap().take() { + if let Some(r) = self.test.lock().take() { self.testing = false; match r { Ok(ms) => s.success(format!("Reachable — {ms} ms")), @@ -336,7 +337,7 @@ impl ProfileEditor { let (slot, svc, ctx) = (self.test.clone(), s.svc.clone(), s.ctx.clone()); std::thread::spawn(move || { let r = svc.test_connection(&host, port).map_err(|e| e.to_string()); - *slot.lock().unwrap() = Some(r); + *slot.lock() = Some(r); ctx.request_repaint(); }); } diff --git a/rust/app/src/settings.rs b/rust/app/src/settings.rs index 535d62a..617f4e0 100644 --- a/rust/app/src/settings.rs +++ b/rust/app/src/settings.rs @@ -9,7 +9,8 @@ use egui::{RichText, Ui, vec2}; use egui_phosphor::regular as icon; use nexus_core::models::{AppPreferences, KnownHost, Protocol}; use nexus_core::updater::UpdateInfo; -use std::sync::{Arc, Mutex}; +use parking_lot::Mutex; +use std::sync::Arc; /// A result from a worker thread, polled each frame (`Some` while the work is in flight). pub struct Job(Arc>>); @@ -20,14 +21,14 @@ impl Job { let (out, ctx) = (slot.clone(), ctx.clone()); std::thread::spawn(move || { let v = f(); - *out.lock().unwrap() = Some(v); + *out.lock() = Some(v); ctx.request_repaint(); }); Job(slot) } pub fn take(&self) -> Option { - self.0.lock().unwrap().take() + self.0.lock().take() } } diff --git a/rust/app/src/sftp.rs b/rust/app/src/sftp.rs index f4a4c4a..76d7c41 100644 --- a/rust/app/src/sftp.rs +++ b/rust/app/src/sftp.rs @@ -13,9 +13,10 @@ use nexus_core::clients::sftp::{self as rsftp, FileEntry, Progress}; use nexus_core::keyfiles::safe_sftp_upload_name; use nexus_core::models::Protocol; use nexus_core::service::Service; +use parking_lot::Mutex; use std::collections::{BTreeMap, HashMap, HashSet}; use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; /// Worker results, applied on the UI thread. enum Done { @@ -47,7 +48,7 @@ struct Inbox(Arc>>, egui::Context); impl Inbox { fn push(&self, d: Done) { - self.0.lock().unwrap().push(d); + self.0.lock().push(d); self.1.request_repaint(); } } @@ -250,7 +251,7 @@ impl Sftp { let tr = self.transfers.remove(i); if self.open_for.as_deref() != Some(session) { let verb = if tr.upload { "↑ Uploaded" } else { "↓ Downloaded" }; - self.inbox.lock().unwrap().push(Done::Toast(format!("{verb} {}", base_name(&tr.path)))); + self.inbox.lock().push(Done::Toast(format!("{verb} {}", base_name(&tr.path)))); } } } @@ -269,7 +270,7 @@ impl Sftp { } fn drain(&mut self, s: &mut Shared) { - let items = std::mem::take(&mut *self.inbox.lock().unwrap()); + let items = std::mem::take(&mut *self.inbox.lock()); for d in items { match d { Done::Listed { session, path, files } => { diff --git a/rust/app/src/terminal.rs b/rust/app/src/terminal.rs index dbb398c..303e10f 100644 --- a/rust/app/src/terminal.rs +++ b/rust/app/src/terminal.rs @@ -12,7 +12,8 @@ use alacritty_terminal::term::test::TermSize; use alacritty_terminal::term::{Config, Term, TermMode}; use alacritty_terminal::vte::ansi::{self, Color, CursorShape, NamedColor, Processor, StdSyncHandler}; use egui::{Color32, Event, FontId, Key, Modifiers, PointerButton, Pos2, Rect, Sense, Stroke, Ui, pos2, vec2}; -use std::sync::{Arc, Mutex}; +use parking_lot::Mutex; +use std::sync::Arc; #[derive(Clone, Default)] struct Listener(Arc>>); @@ -20,7 +21,7 @@ struct Listener(Arc>>); impl EventListener for Listener { fn send_event(&self, e: TermEvent) { if matches!(e, TermEvent::PtyWrite(_) | TermEvent::Title(_) | TermEvent::Bell | TermEvent::ClipboardStore(..)) { - self.0.lock().unwrap().push(e); + self.0.lock().push(e); } } } @@ -294,7 +295,7 @@ impl TermView { self.mouse(ui, &resp, origin, cw, ch, &mut out); // Terminal-originated events. - for e in std::mem::take(&mut *self.events.0.lock().unwrap()) { + for e in std::mem::take(&mut *self.events.0.lock()) { match e { TermEvent::PtyWrite(s) => out.reply.extend_from_slice(s.as_bytes()), TermEvent::Title(t) => out.title = Some(t), @@ -918,7 +919,7 @@ mod tests { fn replies_to_device_attributes() { let mut t = TermView::new(10, 14.0, CursorStyle::Block); t.feed(b"\x1b[6n"); - let ev = t.events.0.lock().unwrap(); + let ev = t.events.0.lock(); assert!(ev.iter().any(|e| matches!(e, TermEvent::PtyWrite(s) if s == "\x1b[1;1R"))); } diff --git a/rust/app/src/vnc_view.rs b/rust/app/src/vnc_view.rs index e8a911f..756cb63 100644 --- a/rust/app/src/vnc_view.rs +++ b/rust/app/src/vnc_view.rs @@ -218,7 +218,7 @@ impl VncView { fn upload(&mut self, ui: &Ui) { let Some(c) = &self.client else { return }; let fb = c.framebuffer(); - let mut fb = fb.lock().unwrap_or_else(|e| e.into_inner()); + let mut fb = fb.lock(); if !fb.dirty || fb.width == 0 || fb.height == 0 { return; } diff --git a/rust/core/Cargo.toml b/rust/core/Cargo.toml index e5bfeab..e2bbc76 100644 --- a/rust/core/Cargo.toml +++ b/rust/core/Cargo.toml @@ -22,6 +22,7 @@ sha2 = "0.11.0" hmac = "0.13.0" sha1 = "0.11.0" zeroize = "1" +parking_lot = "0.12" portable-pty = "0.9.0" serialport = { version = "4", default-features = false } des = "0.9" diff --git a/rust/core/src/clients/github.rs b/rust/core/src/clients/github.rs index 4c41275..345ca4c 100644 --- a/rust/core/src/clients/github.rs +++ b/rust/core/src/clients/github.rs @@ -355,7 +355,8 @@ fn copy_limited_archive(dst: &mut impl Write, src: impl Read, max_bytes: u64) -> #[cfg(test)] mod tests { use super::*; - use std::sync::{Arc, Mutex}; + use parking_lot::Mutex; + use std::sync::Arc; #[test] fn read_limited_rejects_oversized_response() { @@ -410,7 +411,7 @@ mod tests { std::thread::spawn(move || { for req in server.incoming_requests() { let header = |name: &'static str| req.headers().iter().find(|h| h.field.equiv(name)).map(|h| h.value.to_string()); - log.lock().unwrap().push((req.url().to_string(), header("Authorization"), header("User-Agent"))); + log.lock().push((req.url().to_string(), header("Authorization"), header("User-Agent"))); let (status, body) = handler(req.url()); let _ = req.respond(tiny_http::Response::from_string(body).with_status_code(status)); } @@ -422,7 +423,7 @@ mod tests { fn validate_token_returns_login_and_reports_status() { let (base, seen) = serve(|_| (200, r#"{"login":"octocat"}"#.into())); assert_eq!(Client::new(&base, "tok").validate_token().unwrap(), "octocat"); - let (url, auth, ua) = seen.lock().unwrap()[0].clone(); + let (url, auth, ua) = seen.lock()[0].clone(); assert_eq!(url, "/user"); assert_eq!(auth.as_deref(), Some("Bearer tok")); assert_eq!(ua.as_deref(), Some(USER_AGENT)); @@ -459,7 +460,7 @@ mod tests { } }); let calls = Mutex::new(Vec::new()); - let repos = Client::new(&base, "t").fetch_repos_with_progress(|done, total| calls.lock().unwrap().push((done, total))).unwrap(); + let repos = Client::new(&base, "t").fetch_repos_with_progress(|done, total| calls.lock().push((done, total))).unwrap(); assert_eq!(repos.len(), 101); assert_eq!(repos[0].full_name, "me/r100"); assert_eq!(repos[0].pushed_at, "2025-06-01T12:00:00Z"); @@ -472,12 +473,12 @@ mod tests { assert!(r0.private && r0.description.is_empty() && r0.language.is_empty()); assert!(repos.iter().filter(|r| r.latest_committer.is_empty()).count() == 99); - let mut calls = calls.into_inner().unwrap(); + let mut calls = calls.into_inner(); assert_eq!(calls[0], (0, 101)); calls.sort(); assert_eq!(calls.last(), Some(&(101, 101))); assert_eq!(calls.len(), 102); - assert_eq!(seen.lock().unwrap().iter().filter(|s| s.0.starts_with("/user/repos")).count(), 2); + assert_eq!(seen.lock().iter().filter(|s| s.0.starts_with("/user/repos")).count(), 2); } #[test] @@ -505,7 +506,7 @@ mod tests { assert_eq!(std::fs::read(&dest).unwrap(), b"ZIPDATA"); let err = client.download_repo_archive("me/r", "", &dest).unwrap_err().to_string(); assert_eq!(err, "GitHub API returned HTTP 404: missing"); - assert_eq!(seen.lock().unwrap()[1].0, "/repos/me/r/zipball"); + assert_eq!(seen.lock()[1].0, "/repos/me/r/zipball"); assert!(client.download_repo_archive("bad", "", &dest).is_err()); std::fs::remove_dir_all(dir).unwrap(); } diff --git a/rust/core/src/clients/irc.rs b/rust/core/src/clients/irc.rs index 833ac4b..80b7c2e 100644 --- a/rust/core/src/clients/irc.rs +++ b/rust/core/src/clients/irc.rs @@ -5,11 +5,12 @@ //! optionally wrapped in TLS). It alternates between short reads and draining the outgoing //! queue, so the transport never has to be split. +use parking_lot::{Mutex, MutexGuard}; use std::collections::HashMap; use std::io::{ErrorKind, Read, Write}; use std::net::{TcpStream, ToSocketAddrs}; +use std::sync::Arc; use std::sync::mpsc::{self, Receiver, Sender}; -use std::sync::{Arc, Mutex, MutexGuard}; use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; @@ -197,7 +198,7 @@ impl Client { } fn st(&self) -> MutexGuard<'_, State> { - self.state.lock().unwrap_or_else(|e| e.into_inner()) + self.state.lock() } /// Connects and blocks until registration succeeds (RPL_WELCOME), fails, or times out. @@ -205,7 +206,7 @@ impl Client { let (ready_tx, ready_rx) = mpsc::channel(); let (out_tx, out_rx) = mpsc::channel(); self.st().ready = Some(ready_tx); - *self.out.lock().unwrap_or_else(|e| e.into_inner()) = Some(out_tx); + *self.out.lock() = Some(out_tx); let client = self.clone(); std::thread::spawn(move || client.io_thread(out_rx)); @@ -234,7 +235,7 @@ impl Client { } fn queue(&self, out: Out) -> bool { - let guard = self.out.lock().unwrap_or_else(|e| e.into_inner()); + let guard = self.out.lock(); guard.as_ref().is_some_and(|tx| tx.send(out).is_ok()) } @@ -471,7 +472,7 @@ impl Client { fn io_thread(self: Arc, out_rx: Receiver) { let result = self.open_transport().and_then(|stream| self.run(stream, &out_rx)); - *self.out.lock().unwrap_or_else(|e| e.into_inner()) = None; + *self.out.lock() = None; let (was_closing, was_connected, ready) = { let mut st = self.st(); let r = (st.closing, st.connected, st.ready.take()); @@ -1037,13 +1038,13 @@ mod tests { fn remember_room_membership_tracks_confirmed_channels() { let updates = Arc::new(Mutex::new(Vec::new())); let log = updates.clone(); - let client = Client::new(Config { on_rooms_changed: Arc::new(move |rooms| log.lock().unwrap().push(rooms.join(","))), ..Default::default() }); + let client = Client::new(Config { on_rooms_changed: Arc::new(move |rooms| log.lock().push(rooms.join(","))), ..Default::default() }); client.remember_room_membership("#One", true); client.remember_room_membership("#two", true); client.remember_room_membership("#one", true); client.remember_room_membership("#missing", false); client.remember_room_membership("#ONE", false); - assert_eq!(updates.lock().unwrap().join("|"), "#One|#One,#two|#two"); + assert_eq!(updates.lock().join("|"), "#One|#One,#two|#two"); assert_eq!(client.st().room_names, vec!["#two"]); } @@ -1109,9 +1110,9 @@ mod tests { nick: "me".into(), connect_timeout: Duration::from_secs(5), on_event: Arc::new(move |e| { - let _ = tx.lock().unwrap().send(e); + let _ = tx.lock().send(e); }), - on_close: Arc::new(move || *c.lock().unwrap() += 1), + on_close: Arc::new(move || *c.lock() += 1), ..cfg }); (client, listener, rx, closes) @@ -1224,11 +1225,11 @@ mod tests { client.close(); srv.expect("QUIT :Nexus Terminal disconnected"); let deadline = Instant::now() + Duration::from_secs(5); - while *closes.lock().unwrap() == 0 && Instant::now() < deadline { + while *closes.lock() == 0 && Instant::now() < deadline { std::thread::sleep(Duration::from_millis(10)); } std::thread::sleep(Duration::from_millis(100)); - assert_eq!(*closes.lock().unwrap(), 1); + assert_eq!(*closes.lock(), 1); assert!(client.handle_input("#a", "late").is_err()); } @@ -1236,7 +1237,7 @@ mod tests { fn remote_close_reports_error_and_close() { let errors = Arc::new(Mutex::new(Vec::new())); let log = errors.clone(); - let (client, listener, rx, closes) = harness(Config { on_error: Arc::new(move |e| log.lock().unwrap().push(e)), ..Default::default() }); + let (client, listener, rx, closes) = harness(Config { on_error: Arc::new(move |e| log.lock().push(e)), ..Default::default() }); let c = client.clone(); let connecting = std::thread::spawn(move || c.connect()); let mut srv = accept(&listener); @@ -1248,8 +1249,8 @@ mod tests { drop(srv); assert_eq!(wait_for(&rx, |e| e.kind == "error").text, "connection closed by server"); std::thread::sleep(Duration::from_millis(100)); - assert_eq!(*errors.lock().unwrap(), vec!["connection closed by server"]); - assert_eq!(*closes.lock().unwrap(), 1); + assert_eq!(*errors.lock(), vec!["connection closed by server"]); + assert_eq!(*closes.lock(), 1); } #[test] @@ -1259,7 +1260,7 @@ mod tests { let err = client.connect().unwrap_err().to_string(); assert_eq!(err, "IRC connection timed out after 300ms"); drop(listener); - assert_eq!(*closes.lock().unwrap(), 0); + assert_eq!(*closes.lock(), 0); } #[test] @@ -1273,7 +1274,7 @@ mod tests { nick: "me".into(), connect_timeout: Duration::from_secs(5), connector: Some(Arc::new(move || { - *flag.lock().unwrap() = true; + *flag.lock() = true; let s = TcpStream::connect(addr)?; s.set_read_timeout(Some(Duration::from_millis(20)))?; Ok(Box::new(s) as Box) @@ -1286,7 +1287,7 @@ mod tests { srv.expect("USER me * * me"); srv.send(":srv 001 me :Welcome"); connecting.join().unwrap().unwrap(); - assert!(*used.lock().unwrap()); + assert!(*used.lock()); client.handle_input("", "/quit gone").unwrap(); srv.expect("QUIT gone"); } diff --git a/rust/core/src/clients/local.rs b/rust/core/src/clients/local.rs index 74ae572..782909a 100644 --- a/rust/core/src/clients/local.rs +++ b/rust/core/src/clients/local.rs @@ -2,10 +2,11 @@ use super::Callbacks; use crate::Result; +use parking_lot::Mutex; use portable_pty::{CommandBuilder, MasterPty, PtySize, native_pty_system}; use std::io::{Read, Write}; use std::path::PathBuf; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; pub fn default_shell() -> String { if cfg!(windows) { @@ -36,7 +37,7 @@ impl LocalShell { } pub fn start(self: &Arc) -> Result<()> { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); if st.4 { return Err("shell already closed".into()); } @@ -77,7 +78,7 @@ impl LocalShell { std::thread::Builder::new().name("nexus-local-wait".into()).spawn(move || { let _ = child.wait(); let was_closed = { - let mut st = me.state.lock().unwrap(); + let mut st = me.state.lock(); let was = st.4; st.4 = true; st.0 = None; @@ -94,14 +95,14 @@ impl LocalShell { if data.iter().any(|b| matches!(b, b'\r' | b'\n' | 4)) { self.capture_cwd(); } - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let r = st.0.as_mut().ok_or("shell not running")?; r.writer.write_all(data)?; Ok(()) } pub fn resize(&self, cols: u16, rows: u16) -> Result<()> { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); if cols > 0 { st.2 = cols; } @@ -117,7 +118,7 @@ impl LocalShell { /// Idempotent; does not fire on_close (matches Go: an explicit close is not an exit). pub fn close(&self) { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); if st.4 { return; } @@ -129,13 +130,13 @@ impl LocalShell { /// The last directory the shell was seen in (used to restart in the same place). pub fn working_directory(&self) -> PathBuf { - self.state.lock().unwrap().1.clone() + self.state.lock().1.clone() } fn capture_cwd(&self) { - let pid = self.state.lock().unwrap().0.as_ref().and_then(|r| r.pid); + let pid = self.state.lock().0.as_ref().and_then(|r| r.pid); if let Some(dir) = pid.and_then(process_cwd).filter(|d| d.is_dir()) { - self.state.lock().unwrap().1 = dir; + self.state.lock().1 = dir; } } } @@ -165,22 +166,19 @@ mod tests { let seen: Arc>> = Arc::default(); let closed = Arc::new(Mutex::new(0)); let (s2, c2) = (seen.clone(), closed.clone()); - let cb = Callbacks { - on_data: Arc::new(move |d| s2.lock().unwrap().extend_from_slice(d)), - on_error: Arc::new(|_| {}), - on_close: Arc::new(move || *c2.lock().unwrap() += 1), - }; + let cb = + Callbacks { on_data: Arc::new(move |d| s2.lock().extend_from_slice(d)), on_error: Arc::new(|_| {}), on_close: Arc::new(move || *c2.lock() += 1) }; let sh = LocalShell::new("/bin/sh", Some("/tmp".into()), 80, 24, cb); sh.start().unwrap(); sh.write(b"cd / && echo local-$((2*21))\n").unwrap(); let t = Instant::now(); - while !String::from_utf8_lossy(&seen.lock().unwrap()).contains("local-42") { + while !String::from_utf8_lossy(&seen.lock()).contains("local-42") { assert!(t.elapsed() < Duration::from_secs(5)); std::thread::sleep(Duration::from_millis(20)); } sh.resize(100, 30).unwrap(); sh.write(b"exit\n").unwrap(); - while *closed.lock().unwrap() == 0 { + while *closed.lock() == 0 { assert!(t.elapsed() < Duration::from_secs(5)); std::thread::sleep(Duration::from_millis(20)); } diff --git a/rust/core/src/clients/mod.rs b/rust/core/src/clients/mod.rs index 837e41f..1d098b5 100644 --- a/rust/core/src/clients/mod.rs +++ b/rust/core/src/clients/mod.rs @@ -69,20 +69,20 @@ pub(crate) fn dial(host: &str, port: i64, timeout_secs: i64) -> (String, std::io /// a lock, a reader thread that feeds `on_read`, and Go's error/close semantics (a read error or EOF /// reports `read error: …` then closes; `close` is idempotent and fires `on_close` once). pub(crate) struct TcpConn { - stream: std::sync::Mutex>, + stream: parking_lot::Mutex>, on_error: ErrorFn, on_close: CloseFn, } impl TcpConn { pub(crate) fn new(on_error: ErrorFn, on_close: CloseFn) -> TcpConn { - TcpConn { stream: std::sync::Mutex::new(None), on_error, on_close } + TcpConn { stream: parking_lot::Mutex::new(None), on_error, on_close } } pub(crate) fn connect(self: &Arc, stream: std::net::TcpStream, mut on_read: impl FnMut(&[u8]) + Send + 'static) -> crate::Result<()> { let mut reader = stream.try_clone()?; { - let mut guard = self.stream.lock().unwrap(); + let mut guard = self.stream.lock(); if guard.is_some() { return Err("already connected".into()); } @@ -115,18 +115,18 @@ impl TcpConn { pub(crate) fn write(&self, data: &[u8]) -> crate::Result<()> { use std::io::Write; - let guard = self.stream.lock().unwrap(); + let guard = self.stream.lock(); let mut s = guard.as_ref().ok_or("not connected")?; s.write_all(data)?; Ok(()) } pub(crate) fn is_connected(&self) -> bool { - self.stream.lock().unwrap().is_some() + self.stream.lock().is_some() } pub(crate) fn close(&self) { - let taken = self.stream.lock().unwrap().take(); + let taken = self.stream.lock().take(); if let Some(s) = taken { let _ = s.shutdown(std::net::Shutdown::Both); (self.on_close)(); diff --git a/rust/core/src/clients/raw.rs b/rust/core/src/clients/raw.rs index f289c5d..3692301 100644 --- a/rust/core/src/clients/raw.rs +++ b/rust/core/src/clients/raw.rs @@ -2,8 +2,9 @@ //! of sent ("send") and received ("recv") frames, JSON capture, and the hex/escape encoders the //! RAW view reuses. +use parking_lot::Mutex; use std::collections::VecDeque; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use base64::Engine; use serde::Serialize; @@ -115,11 +116,11 @@ impl RawClient { } pub fn history(&self) -> Vec { - self.inner.history.lock().unwrap().entries.iter().cloned().collect() + self.inner.history.lock().entries.iter().cloned().collect() } pub fn clear_history(&self) { - *self.inner.history.lock().unwrap() = History::default(); + *self.inner.history.lock() = History::default(); } /// Writes the history to `path` as indented JSON (mode 0600 on Unix), as Go did. @@ -146,7 +147,7 @@ impl Inner { if entry.data.len() > MAX_HISTORY_BYTES { entry.data.drain(..entry.data.len() - MAX_HISTORY_BYTES); } - let mut h = self.history.lock().unwrap(); + let mut h = self.history.lock(); h.bytes += entry.data.len(); h.entries.push_back(entry); while h.entries.len() > MAX_HISTORY_ENTRIES || h.bytes > MAX_HISTORY_BYTES { diff --git a/rust/core/src/clients/serial.rs b/rust/core/src/clients/serial.rs index bee042f..14ff7b2 100644 --- a/rust/core/src/clients/serial.rs +++ b/rust/core/src/clients/serial.rs @@ -2,8 +2,9 @@ //! `Profile::host`, line settings come from `Profile::serial`. Same conventions as the other //! clients: blocking `connect`, an own reader thread, `on_close` exactly once (unplug included). +use parking_lot::Mutex; +use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; use std::time::Duration; use serialport::{DataBits, FlowControl, Parity, SerialPort, StopBits}; @@ -45,7 +46,7 @@ impl SerialClient { /// Opens and configures the port, applies DTR/RTS, and starts the reader thread. pub fn connect(&self) -> crate::Result<()> { let inner = &self.inner; - if inner.port.lock().unwrap().is_some() { + if inner.port.lock().is_some() { return Err("already connected".into()); } if inner.path.is_empty() { @@ -82,7 +83,7 @@ impl SerialClient { reader.set_timeout(READ_POLL)?; let stop = Arc::new(AtomicBool::new(false)); { - let mut guard = inner.port.lock().unwrap(); + let mut guard = inner.port.lock(); if guard.is_some() { return Err("already connected".into()); } @@ -128,11 +129,11 @@ impl SerialClient { } pub fn is_connected(&self) -> bool { - self.inner.port.lock().unwrap().is_some() + self.inner.port.lock().is_some() } fn with_port(&self, f: impl FnOnce(&mut Box) -> crate::Result<()>) -> crate::Result<()> { - let mut guard = self.inner.port.lock().unwrap(); + let mut guard = self.inner.port.lock(); let (port, _) = guard.as_mut().ok_or("not connected")?; f(port) } @@ -165,7 +166,7 @@ impl Inner { } fn close(&self) { - let taken = self.port.lock().unwrap().take(); + let taken = self.port.lock().take(); if let Some((port, stop)) = taken { stop.store(true, Ordering::SeqCst); drop(port); diff --git a/rust/core/src/clients/ssh.rs b/rust/core/src/clients/ssh.rs index 29b434b..aee14d3 100644 --- a/rust/core/src/clients/ssh.rs +++ b/rust/core/src/clients/ssh.rs @@ -6,11 +6,12 @@ use super::{Callbacks, rt}; use crate::Result; use crate::models::{ForwardDirection, JumpHost, PortForward, Profile, SshConfig}; use base64::Engine; +use parking_lot::Mutex; use russh::client::{self, Handle, Msg}; use russh::keys::{HashAlg, PrivateKeyWithHashAlg, PublicKey}; use russh::{Channel, ChannelMsg, Pty}; use std::collections::HashMap; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::sync::mpsc; @@ -75,7 +76,7 @@ impl client::Handler for Handler { match res { Ok(()) => Ok(true), Err(why) => { - *self.shared.rejection.lock().unwrap() = Some(why); + *self.shared.rejection.lock() = Some(why); Ok(false) } } @@ -91,7 +92,7 @@ impl client::Handler for Handler { reply: client::ChannelOpenHandle, _session: &mut client::Session, ) -> std::result::Result<(), Self::Error> { - let dest = self.shared.remote_targets.lock().unwrap().get(&(connected_address.to_string(), connected_port)).cloned(); + let dest = self.shared.remote_targets.lock().get(&(connected_address.to_string(), connected_port)).cloned(); let Some((host, port)) = dest else { reply.reject(russh::ChannelOpenFailure::AdministrativelyProhibited).await; return Ok(()); @@ -485,7 +486,7 @@ impl SshClient { match tokio::time::timeout(timeout, client::connect_stream(config(&self.sc()), stream, handler)).await { Err(_) => Err(format!("SSH handshake with {host}:{port} timed out").into()), Ok(Ok(h)) => Ok(h), - Ok(Err(e)) => match self.shared.rejection.lock().unwrap().take() { + Ok(Err(e)) => match self.shared.rejection.lock().take() { Some(why) => Err(why.into()), None => Err(format!("SSH handshake failed: {e}").into()), }, @@ -518,15 +519,15 @@ impl SshClient { .channel_open_direct_tcpip(p.host.clone(), port as u32, "127.0.0.1", 0) .await .map_err(|e| format!("failed to dial target {}:{port} through jump hosts: {e}", p.host))?; - self.hops.lock().unwrap().extend(h); - self.hops.lock().unwrap().push(prev); + self.hops.lock().extend(h); + self.hops.lock().push(prev); self.handshake(ch.into_stream(), &p.host, port, &check).await? }; authenticate(&mut handle, Auth { user: &p.username, methods, agent: sc.agent_auth, key, password, totp_secret: sc.totp_secret.clone() }).await?; let handle = Arc::new(handle); if !shell { - *self.handle.lock().unwrap() = Some(handle); + *self.handle.lock() = Some(handle); return Ok(()); } @@ -551,8 +552,8 @@ impl SshClient { let (mut read, write) = ch.split(); let (tx, mut rx) = mpsc::unbounded_channel::(); - *self.tx.lock().unwrap() = Some(tx); - *self.handle.lock().unwrap() = Some(handle); + *self.tx.lock() = Some(tx); + *self.handle.lock() = Some(handle); tokio::spawn(async move { while let Some(cmd) = rx.recv().await { let r = match cmd { @@ -633,16 +634,16 @@ impl SshClient { } fn handle(&self) -> Result>> { - self.handle.lock().unwrap().clone().ok_or_else(|| "not connected".into()) + self.handle.lock().clone().ok_or_else(|| "not connected".into()) } pub fn write(&self, data: &[u8]) -> Result<()> { - let tx = self.tx.lock().unwrap(); + let tx = self.tx.lock(); tx.as_ref().ok_or("not connected")?.send(Cmd::Data(data.to_vec())).map_err(|_| "not connected".into()) } pub fn resize(&self, cols: u32, rows: u32) -> Result<()> { - let tx = self.tx.lock().unwrap(); + let tx = self.tx.lock(); tx.as_ref().ok_or("not connected")?.send(Cmd::Resize(cols, rows)).map_err(|_| "not connected".into()) } @@ -679,7 +680,7 @@ impl SshClient { } pub fn is_connected(&self) -> bool { - self.handle.lock().unwrap().as_ref().is_some_and(|h| !h.is_closed()) + self.handle.lock().as_ref().is_some_and(|h| !h.is_closed()) } /// Idempotent; fires on_close once. @@ -688,9 +689,9 @@ impl SshClient { return; } self.stop_all_forwards(); - *self.tx.lock().unwrap() = None; - let h = self.handle.lock().unwrap().take(); - let hops = std::mem::take(&mut *self.hops.lock().unwrap()); + *self.tx.lock() = None; + let h = self.handle.lock().take(); + let hops = std::mem::take(&mut *self.hops.lock()); rt().spawn(async move { if let Some(h) = h { let _ = h.disconnect(russh::Disconnect::ByApplication, "", "en").await; @@ -705,7 +706,7 @@ impl SshClient { // --- port forwarding ------------------------------------------------------------------- pub fn start_forward(&self, rule: PortForward) -> Result<()> { - if self.forwards.lock().unwrap().contains_key(&rule.id) { + if self.forwards.lock().contains_key(&rule.id) { return Err(format!("forward {:?} is already active", rule.id).into()); } let h = self.handle()?; @@ -727,12 +728,12 @@ impl SshClient { let _ = tokio::io::copy_bidirectional(&mut sock, &mut remote).await; } }); - let mut handles = t2.lock().unwrap(); + let mut handles = t2.lock(); handles.retain(|h| !h.is_finished()); handles.push(conn.abort_handle()); } }); - tasks.lock().unwrap().push(task.abort_handle()); + tasks.lock().push(task.abort_handle()); } ForwardDirection::Remote => { let bind = if rule.bind_host.is_empty() { "0.0.0.0".to_string() } else { rule.bind_host.clone() }; @@ -740,26 +741,26 @@ impl SshClient { .block_on(h.tcpip_forward(bind.clone(), rule.bind_port as u32)) .map_err(|e| format!("remote forward: server listen on {bind}:{} failed: {e}", rule.bind_port))?; let port = if rule.bind_port == 0 { port } else { rule.bind_port as u32 }; - self.shared.remote_targets.lock().unwrap().insert((bind, port), (rule.dest_host.clone(), rule.dest_port as u16)); + self.shared.remote_targets.lock().insert((bind, port), (rule.dest_host.clone(), rule.dest_port as u16)); } } - self.forwards.lock().unwrap().insert(rule.id.clone(), Forward { rule, tasks }); + self.forwards.lock().insert(rule.id.clone(), Forward { rule, tasks }); Ok(()) } pub fn stop_forward(&self, id: &str) -> Result<()> { - let f = self.forwards.lock().unwrap().remove(id).ok_or_else(|| format!("forward {id:?} not found"))?; + let f = self.forwards.lock().remove(id).ok_or_else(|| format!("forward {id:?} not found"))?; self.shutdown_forward(f); Ok(()) } fn shutdown_forward(&self, f: Forward) { - for t in f.tasks.lock().unwrap().drain(..) { + for t in f.tasks.lock().drain(..) { t.abort(); } if f.rule.direction == ForwardDirection::Remote { let bind = if f.rule.bind_host.is_empty() { "0.0.0.0".to_string() } else { f.rule.bind_host.clone() }; - self.shared.remote_targets.lock().unwrap().retain(|(a, p), _| !(a == &bind && *p == f.rule.bind_port as u32)); + self.shared.remote_targets.lock().retain(|(a, p), _| !(a == &bind && *p == f.rule.bind_port as u32)); if let Ok(h) = self.handle() { let port = f.rule.bind_port as u32; rt().spawn(async move { @@ -770,14 +771,14 @@ impl SshClient { } fn stop_all_forwards(&self) { - let all: Vec = self.forwards.lock().unwrap().drain().map(|(_, f)| f).collect(); + let all: Vec = self.forwards.lock().drain().map(|(_, f)| f).collect(); for f in all { self.shutdown_forward(f); } } pub fn active_forwards(&self) -> Vec { - self.forwards.lock().unwrap().values().map(|f| f.rule.clone()).collect() + self.forwards.lock().values().map(|f| f.rule.clone()).collect() } // --- SFTP ------------------------------------------------------------------------------- diff --git a/rust/core/src/clients/telnet.rs b/rust/core/src/clients/telnet.rs index 998b079..0271491 100644 --- a/rust/core/src/clients/telnet.rs +++ b/rust/core/src/clients/telnet.rs @@ -2,7 +2,8 @@ //! TTYPE, NAWS; everything else refused), IAC escaping, line-ending conversion, ISO 8859-1 output //! encoding and BRK. Go read but never acted on `lineMode`, and neither do we. -use std::sync::{Arc, Mutex}; +use parking_lot::Mutex; +use std::sync::Arc; use super::{Callbacks, DataFn, DebugFn, TcpConn}; use crate::models::{LineEnding, Profile, TelnetConfig}; @@ -96,7 +97,7 @@ impl TelnetClient { return Err("not connected".into()); } { - let mut st = self.inner.state.lock().unwrap(); + let mut st = self.inner.state.lock(); st.cols = cols; st.rows = rows; } @@ -119,7 +120,7 @@ impl TelnetClient { /// The negotiated local-echo state: starts at the profile's `localEcho`, cleared by the /// server's WILL ECHO and set by WONT ECHO. Like Go, the client itself never echoes. pub fn local_echo(&self) -> bool { - self.inner.state.lock().unwrap().local_echo + self.inner.state.lock().local_echo } } @@ -196,7 +197,7 @@ impl Inner { OPT_NAWS => { self.negotiate(WILL, opt); let (cols, rows) = { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); (st.cols, st.rows) }; let _ = self.send_naws(cols, rows); @@ -215,7 +216,7 @@ impl Inner { match opt { OPT_ECHO => { self.negotiate(DO, opt); - self.state.lock().unwrap().local_echo = false; + self.state.lock().local_echo = false; } OPT_SGA => self.negotiate(DO, opt), _ => self.negotiate(DONT, opt), @@ -225,7 +226,7 @@ impl Inner { fn handle_wont(&self, opt: u8) { self.debug(format!("received WONT {opt}")); if opt == OPT_ECHO { - self.state.lock().unwrap().local_echo = true; + self.state.lock().local_echo = true; } self.negotiate(DONT, opt); } @@ -362,7 +363,7 @@ mod tests { on_error: Arc::new(move |e| etx.send(e).unwrap()), on_close: Arc::new(move || ctx.send(()).unwrap()), }; - let client = TelnetClient::new(&profile, cb, Some(Arc::new(move |m| dbg.lock().unwrap().push(m)))); + let client = TelnetClient::new(&profile, cb, Some(Arc::new(move |m| dbg.lock().push(m)))); client.connect().unwrap(); let (mut srv, _) = listener.accept().unwrap(); srv.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); @@ -399,7 +400,7 @@ mod tests { client.write(b"ls\n\xff").unwrap(); assert_eq!(read_n(&mut srv, 6), b"ls\r\n\xff\xff"); - let log = debug.lock().unwrap().clone(); + let log = debug.lock().clone(); for m in ["received DO 24", "sending WILL 31", "sending NAWS: 80x24", "sending WONT 99", "sending terminal type: xterm-256color"] { assert!(log.iter().any(|l| l == m), "missing debug {m:?} in {log:?}"); } diff --git a/rust/core/src/clients/vnc.rs b/rust/core/src/clients/vnc.rs index 9567966..8d1c1e7 100644 --- a/rust/core/src/clients/vnc.rs +++ b/rust/core/src/clients/vnc.rs @@ -6,9 +6,10 @@ //! the shared `Arc>` the UI reads, so the lock is never held across network reads. //! Outgoing input is queued and written whenever a read polls out, so the transport is never split. +use parking_lot::{Mutex, MutexGuard}; use std::io::{BufReader, ErrorKind, Read, Write}; +use std::sync::Arc; use std::sync::mpsc::{self, Receiver, Sender, TryRecvError}; -use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; use flate2::{Decompress, FlushDecompress}; @@ -152,7 +153,7 @@ impl Client { } fn st(&self) -> MutexGuard<'_, State> { - self.state.lock().unwrap_or_else(|e| e.into_inner()) + self.state.lock() } /// Connects and blocks until the handshake completes (ServerInit received), fails, or times out. @@ -160,7 +161,7 @@ impl Client { let (ready_tx, ready_rx) = mpsc::channel(); let (out_tx, out_rx) = mpsc::channel(); self.st().ready = Some(ready_tx); - *self.out.lock().unwrap_or_else(|e| e.into_inner()) = Some(out_tx); + *self.out.lock() = Some(out_tx); let client = self.clone(); std::thread::spawn(move || client.io_thread(out_rx)); @@ -186,7 +187,7 @@ impl Client { st.closing = true; } // Dropping the sender makes the I/O thread's next poll end the connection. - *self.out.lock().unwrap_or_else(|e| e.into_inner()) = None; + *self.out.lock() = None; } pub fn framebuffer(&self) -> Arc> { @@ -233,20 +234,20 @@ impl Client { /// Asks for a non-incremental update of the whole screen. pub fn request_full_update(&self) -> crate::Result<()> { let (w, h) = { - let fb = self.fb.lock().unwrap_or_else(|e| e.into_inner()); + let fb = self.fb.lock(); (fb.width, fb.height) }; self.queue(update_request(false, w, h)) } fn queue(&self, msg: Vec) -> crate::Result<()> { - let guard = self.out.lock().unwrap_or_else(|e| e.into_inner()); + let guard = self.out.lock(); if guard.as_ref().is_some_and(|tx| tx.send(msg).is_ok()) { Ok(()) } else { Err("not connected".into()) } } fn io_thread(self: Arc, out_rx: Receiver>) { let result = self.open_transport().and_then(|io| self.run(io, out_rx)); - *self.out.lock().unwrap_or_else(|e| e.into_inner()) = None; + *self.out.lock() = None; let (was_closing, was_connected, ready) = { let mut st = self.st(); let r = (st.closing, st.connected, st.ready.take()); @@ -283,7 +284,7 @@ impl Client { let mut r = BufReader::with_capacity(1 << 16, Conn { io, out_rx }); let (w, h) = self.handshake(&mut r)?; let mut local = Framebuffer::new_checked(w, h)?; - *self.fb.lock().unwrap_or_else(|e| e.into_inner()) = local.clone(); + *self.fb.lock() = local.clone(); (self.cfg.on_resize)(w, h); { let mut st = self.st(); @@ -302,7 +303,7 @@ impl Client { 0 => { let (y0, y1, resized) = read_update(&mut r, &mut local, &mut dec)?; { - let mut fb = self.fb.lock().unwrap_or_else(|e| e.into_inner()); + let mut fb = self.fb.lock(); if resized { *fb = local.clone(); } else if y0 < y1 { @@ -983,7 +984,7 @@ mod tests { frame_rx.recv_timeout(Duration::from_secs(5)).unwrap(); { let fb = client.framebuffer(); - let fb = fb.lock().unwrap(); + let fb = fb.lock(); assert_eq!((fb.width, fb.height), (4, 4)); assert!(fb.dirty); let px = |x: usize, y: usize| &fb.rgba[(y * 4 + x) * 4..(y * 4 + x) * 4 + 4]; diff --git a/rust/core/src/recording.rs b/rust/core/src/recording.rs index 32a93e1..5d2180b 100644 --- a/rust/core/src/recording.rs +++ b/rust/core/src/recording.rs @@ -3,12 +3,12 @@ use crate::Result; use crate::models::Time; +use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::fs::File; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; -use std::sync::Mutex; use std::time::Instant; pub const EVENT_OUTPUT: &str = "o"; @@ -156,7 +156,7 @@ impl Recorder { } fn write_event(&self, kind: &str, data: &[u8]) { - let mut g = self.inner.lock().unwrap(); + let mut g = self.inner.lock(); if g.closed || g.file.is_none() { return; } @@ -177,7 +177,7 @@ impl Recorder { /// Flushes and closes the file. Idempotent. pub fn close(&self) -> Result<()> { - let mut g = self.inner.lock().unwrap(); + let mut g = self.inner.lock(); if g.closed { return Ok(()); } @@ -198,7 +198,7 @@ impl Recorder { } pub fn event_count(&self) -> usize { - self.inner.lock().unwrap().event_count + self.inner.lock().event_count } } diff --git a/rust/core/src/secrets.rs b/rust/core/src/secrets.rs index 85f9ecb..2a13a72 100644 --- a/rust/core/src/secrets.rs +++ b/rust/core/src/secrets.rs @@ -6,11 +6,11 @@ use aes_gcm::aead::{Aead, KeyInit}; use aes_gcm::{Aes256Gcm, Nonce}; use base64::Engine; use base64::engine::general_purpose::STANDARD as B64; +use parking_lot::Mutex; use rand::RngCore; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::path::PathBuf; -use std::sync::Mutex; use zeroize::Zeroizing; const VAULT_FILE: &str = "vault.json"; @@ -93,7 +93,7 @@ impl Vault { } pub fn is_unlocked(&self) -> bool { - self.state.lock().unwrap().is_some() + self.state.lock().is_some() } pub fn initialise(&self, password: &str) -> Result<()> { @@ -104,7 +104,7 @@ impl Vault { let u = Unlocked { key: derive_key(password, &salt)?, vault: VaultFile { version: 1, salt: B64.encode(salt), secrets: BTreeMap::new() } }; std::fs::create_dir_all(&self.dir)?; self.save(&u)?; - *self.state.lock().unwrap() = Some(u); + *self.state.lock() = Some(u); Ok(()) } @@ -121,12 +121,12 @@ impl Vault { let data = B64.decode(v).unwrap_or_default(); open(&u.key, &data).map_err(|_| "secrets: incorrect master password or corrupt vault")?; } - *self.state.lock().unwrap() = Some(u); + *self.state.lock() = Some(u); Ok(()) } pub fn lock(&self) { - *self.state.lock().unwrap() = None; + *self.state.lock() = None; } fn save(&self, u: &Unlocked) -> Result<()> { @@ -135,7 +135,7 @@ impl Vault { } pub fn get(&self, key: &str) -> Result { - let st = self.state.lock().unwrap(); + let st = self.state.lock(); let u = st.as_ref().ok_or("secrets: vault is locked")?; let enc = u.vault.secrets.get(key).ok_or_else(|| format!("secrets: key {key:?} not found"))?; let data = B64.decode(enc).map_err(|e| format!("failed to decode secret: {e}"))?; @@ -143,7 +143,7 @@ impl Vault { } pub fn set(&self, key: &str, value: &str) -> Result<()> { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let u = st.as_mut().ok_or("secrets: vault is locked")?; let enc = B64.encode(seal(&u.key, value.as_bytes())); u.vault.secrets.insert(key.to_string(), enc); @@ -151,14 +151,14 @@ impl Vault { } pub fn delete_prefix(&self, prefix: &str) -> Result<()> { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let u = st.as_mut().ok_or("secrets: vault is locked")?; u.vault.secrets.retain(|k, _| !k.starts_with(prefix)); self.save(u) } pub fn delete(&self, key: &str) -> Result<()> { - let mut st = self.state.lock().unwrap(); + let mut st = self.state.lock(); let u = st.as_mut().ok_or("secrets: vault is locked")?; u.vault.secrets.remove(key); self.save(u) diff --git a/rust/core/src/sessions.rs b/rust/core/src/sessions.rs index 6ccf875..ff7c515 100644 --- a/rust/core/src/sessions.rs +++ b/rust/core/src/sessions.rs @@ -10,11 +10,12 @@ use crate::models::*; use crate::recording::{self, Recorder}; use crate::secrets::Secrets; use crate::storage::Store; +use parking_lot::Mutex; use std::collections::{HashMap, HashSet}; use std::path::PathBuf; +use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc; -use std::sync::{Arc, Mutex}; use std::time::Duration; /// Everything the UI hears about. Prompts carry their reply channel; dropping it cancels. @@ -222,7 +223,7 @@ struct SecretState { /// After a failed attempt, typed credentials are dropped and, when the server refused them, the /// saved ones used are skipped next time so the user can enter a replacement. fn attempt_failed(state: &SessionSecrets, error: &str) { - let mut st = state.lock().unwrap(); + let mut st = state.lock(); st.pending.clear(); if is_auth_error(error) { let used = std::mem::take(&mut st.used); @@ -263,11 +264,11 @@ impl Manager { } pub fn set_sink(&self, sink: EventSink) { - *self.sink.lock().unwrap() = Some(sink); + *self.sink.lock() = Some(sink); } fn emit(&self, e: Event) { - let sink = self.sink.lock().unwrap().clone(); + let sink = self.sink.lock().clone(); if let Some(s) = sink { s(e); } @@ -279,11 +280,11 @@ impl Manager { /// True while `id` still names the Session object with `epoch` and it isn't closed. fn is_current(&self, id: &str, epoch: u64) -> bool { - self.sessions.lock().unwrap().get(id).is_some_and(|s| s.epoch == epoch && !s.closed) + self.sessions.lock().get(id).is_some_and(|s| s.epoch == epoch && !s.closed) } fn with(&self, id: &str, f: impl FnOnce(&mut Session) -> R) -> Result { - let mut map = self.sessions.lock().unwrap(); + let mut map = self.sessions.lock(); match map.get_mut(id) { Some(s) => Ok(f(s)), None => err(format!("sessions: session {id:?} not found")), @@ -312,7 +313,7 @@ impl Manager { session_secrets, }; let id = s.info.id.clone(); - self.sessions.lock().unwrap().insert(id, s); + self.sessions.lock().insert(id, s); epoch } @@ -353,7 +354,7 @@ impl Manager { } fn finish_connect(&self, id: &str, epoch: u64, r: Result, on_ok: impl FnOnce(&Self)) { - let mut map = self.sessions.lock().unwrap(); + let mut map = self.sessions.lock(); let current = map.get_mut(id).filter(|s| s.epoch == epoch && !s.closed); match (r, current) { (Ok(client), None) => { @@ -369,7 +370,7 @@ impl Manager { self.emit_state(id, SessionState::Error, &e.to_string()); } (Ok(client), Some(s)) => { - let typed = std::mem::take(&mut s.session_secrets.lock().unwrap().pending); + let typed = std::mem::take(&mut s.session_secrets.lock().pending); s.client = client.clone(); s.info.state = SessionState::Connected; s.info.connected_at = now(); @@ -392,7 +393,7 @@ impl Manager { fn connect_protocol(self: &Arc, id: &str, epoch: u64, p: &Profile, prompt: Prompt) -> Result { let _ = self.with(id, |s| { - let mut st = s.session_secrets.lock().unwrap(); + let mut st = s.session_secrets.lock(); st.pending.clear(); st.used.clear(); }); @@ -431,15 +432,15 @@ impl Manager { /// Callbacks for terminal-style clients: output to the UI, recording and login scripts. fn terminal_callbacks(self: &Arc, id: &str, epoch: u64) -> Callbacks { - let login = self.sessions.lock().unwrap().get(id).map(|s| s.login.clone()).unwrap_or_default(); + let login = self.sessions.lock().get(id).map(|s| s.login.clone()).unwrap_or_default(); let (me, sid) = (self.clone(), id.to_string()); let on_data = Arc::new(move |d: &[u8]| { me.emit(Event::Data { session: sid.clone(), data: d.to_vec() }); - let rec = me.sessions.lock().unwrap().get(&sid).and_then(|s| s.recorder.clone()); + let rec = me.sessions.lock().get(&sid).and_then(|s| s.recorder.clone()); if let Some(r) = rec { r.write_output(d); } - let mut l = login.lock().unwrap(); + let mut l = login.lock(); if l.active { l.buf.extend_from_slice(d); if l.buf.len() > 16384 { @@ -485,10 +486,10 @@ impl Manager { let key = format!("profile:{profile_id}:{kind}"); let (secrets, store) = (self.secrets.clone(), self.store.clone()); let (pid, kind) = (profile_id.to_string(), kind.to_string()); - let state: SessionSecrets = self.sessions.lock().unwrap().get(session).map(|s| s.session_secrets.clone()).unwrap_or_default(); + let state: SessionSecrets = self.sessions.lock().get(session).map(|s| s.session_secrets.clone()).unwrap_or_default(); Arc::new(move || { if mode == SecretMode::Saved { - let mut st = state.lock().unwrap(); + let mut st = state.lock(); if let Some(v) = st.pending.get(&key) { return Ok(v.clone()); } @@ -501,12 +502,12 @@ impl Manager { } } if mode == SecretMode::SessionOnly - && let Some(v) = state.lock().unwrap().session_only.get(&key).cloned() + && let Some(v) = state.lock().session_only.get(&key).cloned() { return Ok(v); } let v = prompt(&format!("Enter {kind}:"))?; - let mut st = state.lock().unwrap(); + let mut st = state.lock(); match mode { SecretMode::Saved => { st.pending.insert(key.clone(), v.clone()); @@ -693,7 +694,7 @@ impl Manager { let retry = auth.then(|| prompt("GitHub token did not work. Please enter a valid token:").ok()).flatten().filter(|t| !t.is_empty()); let Some(t) = retry else { return err(format!("GitHub authentication failed: {e}")) }; if gh.secret_mode == SecretMode::Saved { - let _ = self.with(id, |s| s.session_secrets.lock().unwrap().pending.insert(format!("profile:{}:github-token", p.id), t.clone())); + let _ = self.with(id, |s| s.session_secrets.lock().pending.insert(format!("profile:{}:github-token", p.id), t.clone())); } client = github::Client::new(&gh.base_url, &t); client.validate_token().map_err(|e| format!("GitHub authentication failed: {e}"))? @@ -963,7 +964,7 @@ impl Manager { /// Explicit close: the tab is gone, so no reconnect. pub fn disconnect(&self, id: &str) -> Result<()> { - let s = self.sessions.lock().unwrap().remove(id).ok_or_else(|| format!("sessions: session {id:?} not found"))?; + let s = self.sessions.lock().remove(id).ok_or_else(|| format!("sessions: session {id:?} not found"))?; self.close_session(s); self.store.add_audit("disconnect", "", id, "Session disconnected"); Ok(()) @@ -972,7 +973,7 @@ impl Manager { /// Closes and re-opens a session from its profile, keeping the tab name and pin; returns the /// new session id. pub fn reconnect(self: &Arc, id: &str) -> Result { - let s = self.sessions.lock().unwrap().remove(id).ok_or_else(|| format!("sessions: session {id:?} not found"))?; + let s = self.sessions.lock().remove(id).ok_or_else(|| format!("sessions: session {id:?} not found"))?; let (profile, tab, pinned, proto, shell) = (s.profile.clone(), s.info.tab_name.clone(), s.info.pinned, s.info.protocol, s.local_shell.clone()); let session_secrets = s.session_secrets.clone(); self.close_session(s); @@ -1001,7 +1002,7 @@ impl Manager { fn handle_closed(self: &Arc, id: &str, epoch: u64) { let profile = { - let mut map = self.sessions.lock().unwrap(); + let mut map = self.sessions.lock(); let Some(s) = map.get_mut(id).filter(|s| s.epoch == epoch && !s.closed) else { return }; s.info.state = SessionState::Disconnected; if !matches!(s.client, Client::Local(_)) { @@ -1109,7 +1110,7 @@ impl Manager { } pub fn close_all(&self) { - let all: Vec = self.sessions.lock().unwrap().drain().map(|(_, s)| s).collect(); + let all: Vec = self.sessions.lock().drain().map(|(_, s)| s).collect(); for s in all { self.close_session(s); } @@ -1122,7 +1123,7 @@ impl Manager { return; } let Ok(login) = self.with(id, |s| s.login.clone()) else { return }; - *login.lock().unwrap() = LoginBuf { active: true, buf: vec![] }; + *login.lock() = LoginBuf { active: true, buf: vec![] }; let (me, id) = (self.clone(), id.to_string()); std::thread::spawn(move || { std::thread::sleep(Duration::from_millis(500)); @@ -1157,7 +1158,7 @@ impl Manager { _ => {} } } - *login.lock().unwrap() = LoginBuf::default(); + *login.lock() = LoginBuf::default(); }); } @@ -1202,15 +1203,15 @@ impl Manager { } pub fn info(&self, id: &str) -> Option { - self.sessions.lock().unwrap().get(id).map(|s| s.info.clone()) + self.sessions.lock().get(id).map(|s| s.info.clone()) } pub fn list(&self) -> Vec { - self.sessions.lock().unwrap().values().map(|s| s.info.clone()).collect() + self.sessions.lock().values().map(|s| s.info.clone()).collect() } pub fn profile(&self, id: &str) -> Option { - self.sessions.lock().unwrap().get(id).and_then(|s| s.profile.clone()) + self.sessions.lock().get(id).and_then(|s| s.profile.clone()) } pub fn rename(&self, id: &str, name: &str) -> Result<()> { @@ -1426,7 +1427,7 @@ fn wait_for_output(login: &Mutex, pattern: &str, timeout: Duration) -> let deadline = std::time::Instant::now() + timeout; while std::time::Instant::now() < deadline { { - let mut l = login.lock().unwrap(); + let mut l = login.lock(); let text = String::from_utf8_lossy(&l.buf).into_owned(); if let Some(i) = text.find(pattern) { let consumed = text[..i + pattern.len()].len(); @@ -1469,7 +1470,7 @@ mod tests { let w = warnings.clone(); m.set_sink(Arc::new(move |e| { if let Event::Warning { message, .. } = e { - w.lock().unwrap().push(message); + w.lock().push(message); } })); let p = Profile { host: "127.0.0.1".into(), port: 1, ..Profile::new(Protocol::Raw) }; @@ -1494,7 +1495,7 @@ mod tests { let e = m.new_session(info("s0"), Some(p.clone()), String::new(), SessionSecrets::default()); resolve(&m, "s0"); m.finish_connect("s0", e, Ok(Client::None), |_| {}); - assert_eq!(warnings.lock().unwrap().len(), 1); + assert_eq!(warnings.lock().len(), 1); m.secrets.vault.initialise("pw").unwrap(); let e = m.new_session(info("s1"), Some(p.clone()), String::new(), SessionSecrets::default()); @@ -1570,40 +1571,40 @@ mod tests { let (a, pr) = (answers.clone(), prompts.clone()); m.set_sink(Arc::new(move |e| { if let Event::HostKeyVerify { stored, reply, .. } = e { - pr.lock().unwrap().push(stored); - let _ = reply.send(a.lock().unwrap().remove(0)); + pr.lock().push(stored); + let _ = reply.send(a.lock().remove(0)); } })); let info = |fp: &str| HostKeyInfo { host: "h".into(), port: 22, key_type: "ssh-ed25519".into(), fingerprint: fp.into(), public_key_b64: "AAAA".into() }; let stored = |m: &Manager| m.store.get_known_host("h", 22).unwrap().map(|k| k.fingerprint); // Unknown key: accept once records nothing; accept & save records it. let check = m.host_key_check("s", HostKeyPolicy::Ask); - answers.lock().unwrap().push(HostKeyDecision::AcceptOnce); + answers.lock().push(HostKeyDecision::AcceptOnce); check(&info("A")).unwrap(); assert_eq!(stored(&m), None); - answers.lock().unwrap().push(HostKeyDecision::AcceptAndSave); + answers.lock().push(HostKeyDecision::AcceptAndSave); check(&info("A")).unwrap(); assert_eq!(stored(&m).as_deref(), Some("A")); - assert_eq!(prompts.lock().unwrap().as_slice(), [None, None]); + assert_eq!(prompts.lock().as_slice(), [None, None]); // Same key again: no prompt. check(&info("A")).unwrap(); - assert_eq!(prompts.lock().unwrap().len(), 2); + assert_eq!(prompts.lock().len(), 2); // Changed key: reject, accept once (record untouched), accept & update (record replaced). - answers.lock().unwrap().push(HostKeyDecision::Reject); + answers.lock().push(HostKeyDecision::Reject); assert!(check(&info("B")).unwrap_err().contains("has changed")); - answers.lock().unwrap().push(HostKeyDecision::AcceptOnce); + answers.lock().push(HostKeyDecision::AcceptOnce); check(&info("B")).unwrap(); assert_eq!(stored(&m).as_deref(), Some("A")); - answers.lock().unwrap().push(HostKeyDecision::AcceptAndSave); + answers.lock().push(HostKeyDecision::AcceptAndSave); check(&info("B")).unwrap(); assert_eq!(stored(&m).as_deref(), Some("B")); - assert_eq!(prompts.lock().unwrap()[2..], [Some("A".to_string()), Some("A".into()), Some("A".into())]); + assert_eq!(prompts.lock()[2..], [Some("A".to_string()), Some("A".into()), Some("A".into())]); assert_eq!(m.store.list_known_hosts().unwrap().len(), 1); // Strict and accept-new never prompt on a changed key. for policy in [HostKeyPolicy::Strict, HostKeyPolicy::AcceptNew] { assert!(m.host_key_check("s", policy)(&info("C")).unwrap_err().contains("has changed")); } - assert_eq!(prompts.lock().unwrap().len(), 5); + assert_eq!(prompts.lock().len(), 5); let _ = std::fs::remove_dir_all(dir); } } diff --git a/rust/core/src/storage.rs b/rust/core/src/storage.rs index 6616544..27f5b14 100644 --- a/rust/core/src/storage.rs +++ b/rust/core/src/storage.rs @@ -4,9 +4,9 @@ use crate::Result; use crate::models::*; use chrono::{DateTime, SecondsFormat}; +use parking_lot::Mutex; use rusqlite::{Connection, OptionalExtension, Row, params}; use std::path::Path; -use std::sync::Mutex; const MIGRATIONS: &[(i64, &str)] = &[ (1, include_str!("../migrations/001_initial.sql")), @@ -121,7 +121,7 @@ impl Store { } fn migrate(&self) -> Result<()> { - let db = self.db.lock().unwrap(); + let db = self.db.lock(); db.execute_batch("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL)")?; let current: i64 = db.query_row("SELECT COALESCE(MAX(version), 0) FROM schema_version", [], |r| r.get(0))?; for (ver, sql) in MIGRATIONS { @@ -170,7 +170,6 @@ impl Store { let sql = format!("INSERT INTO profiles ({PROFILE_COLS}) VALUES ({})", vec!["?"; 28].join(",")); self.db .lock() - .unwrap() .execute(&sql, rusqlite::params_from_iter(args.iter().map(|b| b.as_ref()))) .map_err(|e| format!("storage: failed to create profile: {e}"))?; Ok(()) @@ -183,7 +182,6 @@ impl Store { let n = self .db .lock() - .unwrap() .execute( "UPDATE profiles SET display_name=?, description=?, protocol=?, host=?, port=?, username=?, tags=?, folder=?, environment=?, notes=?, favourite=?, last_connected=?, @@ -203,14 +201,13 @@ impl Store { pub fn get_profile(&self, id: &str) -> Result { self.db .lock() - .unwrap() .query_row(&format!("SELECT {PROFILE_COLS} FROM profiles WHERE id = ?"), [id], scan_profile) .optional()? .ok_or_else(|| format!("storage: profile {id:?} not found").into()) } pub fn delete_profile(&self, id: &str) -> Result<()> { - let mut db = self.db.lock().unwrap(); + let mut db = self.db.lock(); let tx = db.transaction()?; tx.execute("DELETE FROM irc_channel_state WHERE profile_id = ?", [id])?; if tx.execute("DELETE FROM profiles WHERE id = ?", [id])? == 0 { @@ -221,14 +218,14 @@ impl Store { } pub fn list_profiles(&self) -> Result> { - let db = self.db.lock().unwrap(); + let db = self.db.lock(); let mut st = db.prepare(&format!("SELECT {PROFILE_COLS} FROM profiles ORDER BY display_name"))?; let rows = st.query_map([], scan_profile)?.collect::>>()?; Ok(rows) } pub fn set_favourite(&self, id: &str, fav: bool) -> Result<()> { - let n = self.db.lock().unwrap().execute("UPDATE profiles SET favourite = ?, updated_at = ? WHERE id = ?", params![fav as i64, fmt_time(&now()), id])?; + let n = self.db.lock().execute("UPDATE profiles SET favourite = ?, updated_at = ? WHERE id = ?", params![fav as i64, fmt_time(&now()), id])?; if n == 0 { return Err(format!("storage: profile {id:?} not found").into()); } @@ -237,7 +234,7 @@ impl Store { pub fn update_last_connected(&self, id: &str) -> Result<()> { let now = fmt_time(&now()); - self.db.lock().unwrap().execute("UPDATE profiles SET last_connected = ?, updated_at = ? WHERE id = ?", params![now, now, id])?; + self.db.lock().execute("UPDATE profiles SET last_connected = ?, updated_at = ? WHERE id = ?", params![now, now, id])?; Ok(()) } @@ -273,7 +270,6 @@ impl Store { Ok(self .db .lock() - .unwrap() .query_row( "SELECT id, host, port, key_type, fingerprint, public_key, first_seen, last_seen, trusted FROM known_hosts WHERE host = ? AND port = ?", @@ -284,7 +280,7 @@ impl Store { } pub fn save_known_host(&self, kh: &KnownHost) -> Result<()> { - self.db.lock().unwrap().execute( + self.db.lock().execute( "INSERT INTO known_hosts (id, host, port, key_type, fingerprint, public_key, first_seen, last_seen, trusted) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(host, port, key_type) DO UPDATE SET @@ -306,7 +302,7 @@ impl Store { } pub fn list_known_hosts(&self) -> Result> { - let db = self.db.lock().unwrap(); + let db = self.db.lock(); let mut st = db.prepare( "SELECT id, host, port, key_type, fingerprint, public_key, first_seen, last_seen, trusted FROM known_hosts ORDER BY host, port", @@ -316,21 +312,21 @@ impl Store { } pub fn delete_known_host(&self, id: &str) -> Result<()> { - if self.db.lock().unwrap().execute("DELETE FROM known_hosts WHERE id = ?", [id])? == 0 { + if self.db.lock().execute("DELETE FROM known_hosts WHERE id = ?", [id])? == 0 { return Err(format!("storage: known host {id:?} not found").into()); } Ok(()) } pub fn add_audit(&self, action: &str, profile_id: &str, session_id: &str, details: &str) { - let _ = self.db.lock().unwrap().execute( + let _ = self.db.lock().execute( "INSERT INTO audit_log (id, timestamp, action, profile_id, session_id, details) VALUES (?, ?, ?, ?, ?, ?)", params![new_id(), fmt_time(&now()), action, profile_id, session_id, details], ); } pub fn list_audit(&self, limit: i64) -> Result> { - let db = self.db.lock().unwrap(); + let db = self.db.lock(); let mut st = db.prepare("SELECT id, timestamp, action, profile_id, session_id, details FROM audit_log ORDER BY timestamp DESC LIMIT ?")?; let rows = st .query_map([limit], |r| { @@ -348,7 +344,7 @@ impl Store { } pub fn irc_last_rooms(&self, profile_id: &str) -> Result> { - let db = self.db.lock().unwrap(); + let db = self.db.lock(); let mut st = db.prepare("SELECT room_name FROM irc_channel_state WHERE profile_id = ? ORDER BY position, room_name")?; let rows = st.query_map([profile_id], |r| r.get(0))?.collect::>>()?; Ok(rows) @@ -360,7 +356,7 @@ impl Store { return Err("storage: IRC channel state requires a profile ID".into()); } let rooms = dedup_rooms(&[rooms]); - let mut db = self.db.lock().unwrap(); + let mut db = self.db.lock(); let tx = db.transaction()?; tx.execute("DELETE FROM irc_channel_state WHERE profile_id = ?", [profile_id])?; let now = fmt_time(&now()); @@ -375,15 +371,14 @@ impl Store { } pub fn preferences(&self) -> AppPreferences { - let v: Option = - self.db.lock().unwrap().query_row("SELECT value FROM preferences WHERE key = '_all'", [], |r| r.get(0)).optional().ok().flatten(); + let v: Option = self.db.lock().query_row("SELECT value FROM preferences WHERE key = '_all'", [], |r| r.get(0)).optional().ok().flatten(); let mut p = v.and_then(|v| serde_json::from_str::(&v).ok()).unwrap_or_default(); p.apply_defaults(); p } pub fn save_preferences(&self, p: &AppPreferences) -> Result<()> { - self.db.lock().unwrap().execute( + self.db.lock().execute( "INSERT INTO preferences (key, value) VALUES ('_all', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value", [serde_json::to_string(p)?], )?; @@ -391,14 +386,14 @@ impl Store { } pub fn list_snippets(&self) -> Result> { - let db = self.db.lock().unwrap(); + let db = self.db.lock(); let mut st = db.prepare("SELECT id, name, command, description, folder, tags, created_at, updated_at FROM snippets ORDER BY name ASC")?; let rows = st.query_map([], scan_snippet)?.collect::>>()?; Ok(rows) } pub fn create_snippet(&self, s: &Snippet) -> Result<()> { - self.db.lock().unwrap().execute( + self.db.lock().execute( "INSERT INTO snippets (id, name, command, description, folder, tags, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", params![s.id, s.name, s.command, s.description, s.folder, serde_json::to_string(&s.tags)?, fmt_time(&s.created_at), fmt_time(&s.updated_at)], )?; @@ -406,7 +401,7 @@ impl Store { } pub fn update_snippet(&self, s: &Snippet) -> Result<()> { - self.db.lock().unwrap().execute( + self.db.lock().execute( "UPDATE snippets SET name=?, command=?, description=?, folder=?, tags=?, updated_at=? WHERE id=?", params![s.name, s.command, s.description, s.folder, serde_json::to_string(&s.tags)?, fmt_time(&s.updated_at), s.id], )?; @@ -414,7 +409,7 @@ impl Store { } pub fn delete_snippet(&self, id: &str) -> Result<()> { - self.db.lock().unwrap().execute("DELETE FROM snippets WHERE id = ?", [id])?; + self.db.lock().execute("DELETE FROM snippets WHERE id = ?", [id])?; Ok(()) } } diff --git a/rust/core/tests/sessions.rs b/rust/core/tests/sessions.rs index 8f05529..abc0e5f 100644 --- a/rust/core/tests/sessions.rs +++ b/rust/core/tests/sessions.rs @@ -4,10 +4,11 @@ use nexus_core::models::{Profile, Protocol, SessionState}; use nexus_core::secrets::{Keychain, Secrets, Vault}; use nexus_core::sessions::{Event, HostKeyDecision, Manager}; use nexus_core::storage::Store; +use parking_lot::Mutex; use std::io::{Read, Write}; use std::net::TcpListener; +use std::sync::Arc; use std::sync::mpsc; -use std::sync::{Arc, Mutex}; use std::time::Duration; fn manager(name: &str) -> (Arc, mpsc::Receiver, std::path::PathBuf) { @@ -20,7 +21,7 @@ fn manager(name: &str) -> (Arc, mpsc::Receiver, std::path::PathB let (tx, rx) = mpsc::channel(); let tx = Mutex::new(tx); m.set_sink(Arc::new(move |e| { - let _ = tx.lock().unwrap().send(e); + let _ = tx.lock().send(e); })); (m, rx, dir) } diff --git a/rust/core/tests/ssh_integration.rs b/rust/core/tests/ssh_integration.rs index b489762..886549e 100644 --- a/rust/core/tests/ssh_integration.rs +++ b/rust/core/tests/ssh_integration.rs @@ -5,11 +5,12 @@ use nexus_core::clients::Callbacks; use nexus_core::clients::sftp; use nexus_core::clients::ssh::{HostKeyCheck, SshClient}; use nexus_core::models::{ForwardDirection, JumpHost, PortForward, Profile, Protocol}; +use parking_lot::Mutex; use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; use std::path::{Path, PathBuf}; use std::process::{Child, Command}; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::{Duration, Instant}; struct Server { @@ -87,7 +88,7 @@ fn callbacks() -> (Callbacks, Seen) { let seen: Seen = Arc::default(); let s2 = seen.clone(); let mut cb = Callbacks::noop(); - cb.on_data = Arc::new(move |d| s2.lock().unwrap().extend_from_slice(d)); + cb.on_data = Arc::new(move |d| s2.lock().extend_from_slice(d)); (cb, seen) } @@ -96,7 +97,7 @@ fn accept_all() -> (HostKeyCheck, Arc>>) { let f2 = fps.clone(); ( Arc::new(move |i| { - f2.lock().unwrap().push(format!("{} {}", i.key_type, i.fingerprint)); + f2.lock().push(format!("{} {}", i.key_type, i.fingerprint)); Ok(()) }), fps, @@ -106,7 +107,7 @@ fn accept_all() -> (HostKeyCheck, Arc>>) { fn wait_for(seen: &Seen, needle: &str) -> bool { let start = Instant::now(); while start.elapsed() < Duration::from_secs(10) { - if String::from_utf8_lossy(&seen.lock().unwrap()).contains(needle) { + if String::from_utf8_lossy(&seen.lock()).contains(needle) { return true; } std::thread::sleep(Duration::from_millis(50)); @@ -160,11 +161,11 @@ fn shell_exec_forwards_sftp() { let (check, fps) = accept_all(); let c = SshClient::new(&profile(&s), cb); c.connect(check, None, None).unwrap(); - assert!(fps.lock().unwrap()[0].starts_with("ssh-ed25519 SHA256:")); + assert!(fps.lock()[0].starts_with("ssh-ed25519 SHA256:")); c.resize(120, 40).unwrap(); c.write(b"echo nexus-$((40+2)) $COLUMNS\n").unwrap(); - assert!(wait_for(&seen, "nexus-42"), "shell output: {}", String::from_utf8_lossy(&seen.lock().unwrap())); + assert!(wait_for(&seen, "nexus-42"), "shell output: {}", String::from_utf8_lossy(&seen.lock())); assert_eq!(c.run_command("echo hi").unwrap(), "hi"); assert!(c.ping().unwrap() >= 0); @@ -222,16 +223,16 @@ fn close_fires_once_and_remote_exit_closes() { let (mut cb, _) = callbacks(); let n = Arc::new(Mutex::new(0)); let n2 = n.clone(); - cb.on_close = Arc::new(move || *n2.lock().unwrap() += 1); + cb.on_close = Arc::new(move || *n2.lock() += 1); let c = SshClient::new(&profile(&s), cb); c.connect(accept_all().0, None, None).unwrap(); c.write(b"exit\n").unwrap(); let start = Instant::now(); - while *n.lock().unwrap() == 0 && start.elapsed() < Duration::from_secs(10) { + while *n.lock() == 0 && start.elapsed() < Duration::from_secs(10) { std::thread::sleep(Duration::from_millis(50)); } c.close(); - assert_eq!(*n.lock().unwrap(), 1); + assert_eq!(*n.lock(), 1); } #[test] @@ -252,7 +253,7 @@ fn jump_host_chain() { let (check, fps) = accept_all(); let c = SshClient::new(&p, Callbacks::noop()); c.connect(check, None, None).unwrap(); - assert_eq!(fps.lock().unwrap().len(), 2, "jump and target both verified"); + assert_eq!(fps.lock().len(), 2, "jump and target both verified"); assert_eq!(c.run_command("echo via-jump").unwrap(), "via-jump"); c.close(); } @@ -281,6 +282,6 @@ fn agent_auth_and_forwarding() { let cl = SshClient::new(&p, cb); cl.connect(accept_all().0, None, None).unwrap(); cl.write(b"ssh-add -l >/dev/null && echo agent-$((1+1))-ok\n").unwrap(); - assert!(wait_for(&seen, "agent-2-ok"), "forwarded agent not usable: {}", String::from_utf8_lossy(&seen.lock().unwrap())); + assert!(wait_for(&seen, "agent-2-ok"), "forwarded agent not usable: {}", String::from_utf8_lossy(&seen.lock())); cl.close(); }