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

CH-J Tools

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

下载源代码 ZIP
CH-J / Server Manager 1704 个文件
0.1.2-beta 添加时间: · 构建时间:
session.rs 添加时间: 未记录日期 55.7 KB · 1099 行
代码 下载文件
1//! SSH session manager (port of `sessions/sessionManager.js`, connection part).
2//!
3//! One SSH connection per session is shared by the terminal, SFTP, metrics, Users, NGINX,
4//! disks and swap services. Host keys are verified against the SHA-256 fingerprints stored
5//! in the vault (same hex format as ssh2 `hostHash: "sha256"` in 0.1.0); unknown or changed
6//! keys are refused and kept as a 5-minute pending confirmation.
8use crate::core::availability::Outcome;
9use crate::core::error::{CoreError, CoreResult};
10use crate::core::logging::Logger;
11use crate::core::profiles::{normalize_profile, ProfileService};
12use crate::core::ssh::exec::{ExecChannel, ExecResult, MAX_COMMAND_OUTPUT};
13use crate::core::ssh::keys::{decode_private_key, read_private_key_file, sha256_hex};
14use crate::core::ssh::utf8::Utf8Decoder;
15use once_cell::sync::Lazy;
16use parking_lot::Mutex;
17use regex::Regex;
18use russh::client::{self, Handle, Msg};
19use russh::keys::{PrivateKeyWithHashAlg, PublicKeyOrCertificate};
20use russh::{ChannelMsg, ChannelWriteHalf};
21use serde_json::{json, Value};
22use std::collections::HashMap;
23use std::net::{IpAddr, SocketAddr};
24use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
25use std::sync::Arc;
26use std::time::{Duration, Instant};
27use tokio::sync::{broadcast, mpsc, watch};
28use zeroize::Zeroizing;
30static SESSION_ID: Lazy<Regex> = Lazy::new(|| Regex::new(r"^[A-Za-z0-9._-]{1,80}$").unwrap());
31pub const MAX_INPUT_LENGTH: usize = 64 * 1024;
32const READY_TIMEOUT: Duration = Duration::from_secs(15);
33const PENDING_TRUST_TTL: Duration = Duration::from_secs(5 * 60);
35pub fn normalize_terminal_size(value: Option<&Value>, fallback: u32, min: u32, max: u32) -> u32 {
36 match value.and_then(Value::as_f64) {
37 Some(n) if n.fract() == 0.0 => (n.max(min as f64).min(max as f64)) as u32,
38 _ => fallback,
39 }
42/// Dropbear's limit for one command (MAX_CMD_LEN); above it the server ends the connection.
43const DROPBEAR_MAX_COMMAND: usize = 8_900;
45pub fn command_too_long_for(server_id: &str, command: &str) -> bool {
46 server_id.to_ascii_lowercase().contains("dropbear") && command.len() > DROPBEAR_MAX_COMMAND
49pub fn shell_quote(value: &str) -> String {
50 format!("'{}'", value.replace('\'', "'\"'\"'"))
53// ---------------------------------------------------------------------------------------------
54// Client handler: host-key verification
55// ---------------------------------------------------------------------------------------------
57pub struct ClientHandler {
58 expected: Option<String>,
59 observed: Arc<Mutex<Option<String>>>,
60 server_id: Arc<Mutex<String>>,
61 negotiated: Arc<Mutex<Value>>,
64impl client::Handler for ClientHandler {
65 type Error = russh::Error;
67 async fn channel_open_confirmation(&mut self, _id: russh::ChannelId, _max_packet_size: u32, _window_size: u32, session: &mut client::Session) -> Result<(), Self::Error> {
68 *self.server_id.lock() = String::from_utf8_lossy(session.remote_sshid()).into_owned();
69 Ok(())
70 }
72 async fn kex_done(&mut self,_shared_secret:Option<&[u8]>,names:&russh::NegotiatedAlgorithms,_session:&mut client::Session)->Result<(),Self::Error>{
73 let mut state=self.negotiated.lock();let count=state["exchanges"].as_u64().unwrap_or(0)+1;
74 let kex=names.kex.as_ref();let host=names.key.as_str();
75 *state=json!({"kex":kex,"hostKeyAlgorithm":host,"kexPostQuantum":kex.starts_with("mlkem") || kex.starts_with("sntrup"),"hostKeyPostQuantum":host=="ssh-mldsa44-ed25519","exchanges":count,"authenticationAlgorithm":state["authenticationAlgorithm"],"authenticationPostQuantum":state["authenticationPostQuantum"]});
76 Ok(())
77 }
78 async fn check_server_key(&mut self, key: &PublicKeyOrCertificate) -> Result<bool, Self::Error> {
79 let blob = match key {
80 PublicKeyOrCertificate::PublicKey { key, .. } => key.to_bytes()?,
81 PublicKeyOrCertificate::Certificate(certificate) => certificate.to_bytes()?,
82 };
83 let fingerprint = sha256_hex(&blob);
84 *self.observed.lock() = Some(fingerprint.clone());
85 Ok(self.expected.as_deref() == Some(fingerprint.as_str()))
86 }
89/// Algorithm lists of ssh2 1.17 defaults used by 0.1.0 (plus ML-KEM hybrid KEX first).
90fn client_config(keepalive: bool) -> Arc<client::Config> {client_config_policy(keepalive,"prefer-hybrid")}
91fn client_config_policy(keepalive:bool,policy:&str)->Arc<client::Config> {
92 use russh::{cipher, kex, mac, Preferred};
93 use std::borrow::Cow;
94 let mut preferred = Preferred {
95 kex: Cow::Owned(vec![
96 kex::MLKEM768X25519_SHA256,
97 kex::MLKEM768NISTP256_SHA256,
98 kex::MLKEM1024NISTP384_SHA384,
99 kex::CURVE25519,
100 kex::CURVE25519_PRE_RFC_8731,
101 kex::ECDH_SHA2_NISTP256,
102 kex::ECDH_SHA2_NISTP384,
103 kex::ECDH_SHA2_NISTP521,
104 kex::DH_GEX_SHA256,
105 kex::DH_G14_SHA256,
106 kex::DH_G15_SHA512,
107 kex::DH_G16_SHA512,
108 kex::DH_G17_SHA512,
109 kex::DH_G18_SHA512,
110 kex::EXTENSION_SUPPORT_AS_CLIENT,
111 kex::EXTENSION_OPENSSH_STRICT_KEX_AS_CLIENT,
112 ]),
113 cipher: Cow::Owned(vec![
114 cipher::CHACHA20_POLY1305,
115 cipher::AES_128_GCM,
116 cipher::AES_256_GCM,
117 cipher::AES_128_CTR,
118 cipher::AES_192_CTR,
119 cipher::AES_256_CTR,
120 ]),
121 mac: Cow::Owned(vec![
122 mac::HMAC_SHA256_ETM,
123 mac::HMAC_SHA512_ETM,
124 mac::HMAC_SHA1_ETM,
125 mac::HMAC_SHA256,
126 mac::HMAC_SHA512,
127 mac::HMAC_SHA1,
128 ]),
129 ..Preferred::DEFAULT
130 };
131 let mut host_algorithms=Preferred::DEFAULT.key.to_vec();
132 host_algorithms.insert(0,russh::keys::ssh_key::Algorithm::new("ssh-mldsa44-ed25519").expect("constant SSH algorithm"));
133 preferred.key=Cow::Owned(host_algorithms);
134 if policy=="hybrid-only"{preferred.kex.to_mut().retain(|name|name.as_ref().starts_with("mlkem") || name.as_ref().starts_with("sntrup") || *name==kex::EXTENSION_SUPPORT_AS_CLIENT || *name==kex::EXTENSION_OPENSSH_STRICT_KEX_AS_CLIENT);}
135 Arc::new(client::Config {
136 preferred,
137 inactivity_timeout: None,
138 keepalive_interval: keepalive.then(|| Duration::from_secs(15)),
139 keepalive_max: 3,
140 nodelay: true,
141 ..Default::default()
142 })
145// ---------------------------------------------------------------------------------------------
146// Session records
147// ---------------------------------------------------------------------------------------------
149enum ShellCommand {
150 Input(Vec<u8>),
151 Resize(u32, u32),
154pub struct SessionRecord {
155 connection_epoch:u64,
156 vault_generation:Option<u64>,
157 pub server_id: Mutex<String>,
158 pub session_id: String,
159 pub profile_id: String,
160 pub label: String,
161 pub host: String,
162 pub address: Mutex<String>,
163 pub port: u16,
164 pub username: String,
165 pub auth_method: String,
166 pub slot: Option<u8>,
167 pub adhoc: bool,
168 profile: Value,
169 route: Vec<Value>,
170 jump_credentials: Mutex<Option<Vec<Credentials>>>,
171 transport: Mutex<Option<OpenedRoute>>,
172 pub state: Mutex<String>,
173 pub handle: Mutex<Option<Arc<Handle<ClientHandler>>>>,
174 shell: Mutex<Option<mpsc::UnboundedSender<ShellCommand>>>,
175 pub latency: Mutex<Option<Value>>,
176 pub finalized: AtomicBool,
177 lifecycle: Mutex<()>,
178 lost_tx: watch::Sender<bool>,
179 pub platform: Mutex<Option<Value>>,
180 pub privilege: Mutex<Option<Value>>,
181 pub sudo_cache: Mutex<Option<(Zeroizing<String>, Instant)>>,
182 pub windows_queue: tokio::sync::Mutex<()>,
183 pub metrics_state: Mutex<Value>,
184 pub stop_latency: Mutex<Option<tokio::task::JoinHandle<()>>>,
187impl SessionRecord {
188 pub fn has_jump(&self) -> bool {
189 self.route.len() > 1
190 }
192 pub fn lost(&self) -> watch::Receiver<bool> {
193 self.lost_tx.subscribe()
194 }
196 pub fn is_connected(&self) -> bool {
197 !self.finalized.load(Ordering::SeqCst) && *self.state.lock() == "connected"
198 }
200 pub fn handle(&self) -> CoreResult<Arc<Handle<ClientHandler>>> {
201 self.handle.lock().clone().ok_or_else(not_connected)
202 }
204 pub fn public_status(&self) -> Value {
205 json!({
206 "sessionId": self.session_id,
207 "profileId": self.profile_id,
208 "label": self.label,
209 "host": self.host,
210 "port": self.port,
211 "username": self.username,
212 "authMethod": self.auth_method,
213 "kexPolicy":self.profile["kexPolicy"].as_str().unwrap_or("prefer-hybrid"),
214 "crypto":self.transport.lock().as_ref().map(|transport|transport.negotiated.iter().map(|v|v.lock().clone()).collect::<Vec<_>>()).unwrap_or_default(),
215 "state": *self.state.lock(),
216 "slot": self.slot,
217 "adhoc": self.adhoc,
218 "jump": self.route.iter().take(self.route.len().saturating_sub(1)).map(|p| json!({
219 "label": p["label"], "host": p["host"], "port": p["port"], "username": p["username"]
220 })).collect::<Vec<_>>(),
221 "latency": self.latency.lock().clone().unwrap_or_else(|| json!({ "pingMs": null, "sshRttMs": null, "measuredAt": null }))
222 })
223 }
226pub fn not_connected() -> CoreError {
227 CoreError::new("SESSION_NOT_CONNECTED", "The terminal session is not connected.")
230#[derive(Clone)]
231struct PendingTrust {
232 profile_id: String,
233 host: String,
234 port: u16,
235 fingerprint: String,
236 known: Option<String>,
237 expires: Instant,
240pub type Emitter = Arc<dyn Fn(&str, Value) + Send + Sync>;
242#[derive(Clone)]
243enum Credentials {
244 Password(Zeroizing<String>),
245 Key(Arc<russh::keys::PrivateKey>),
248// The lease closes every opened transport on errors, cancellation and normal teardown.
249// Reverse order always disconnects the target before the jumps.
250#[derive(Default)]
251struct OpenedRoute {
252 handles: Vec<Arc<Handle<ClientHandler>>>,
253 address: String,
254 /// SSH identification of the last hop, known once a channel is open on it.
255 server_id: Arc<Mutex<String>>,
256 negotiated: Vec<Arc<Mutex<Value>>>,
259struct PendingHandle(Option<Handle<ClientHandler>>);
260impl Drop for PendingHandle {
261 fn drop(&mut self){
262 let Some(handle)=self.0.take()else{return};let Ok(runtime)=tokio::runtime::Handle::try_current()else{return};
263 runtime.spawn(async move{let _=tokio::time::timeout(Duration::from_secs(2),handle.disconnect(russh::Disconnect::ByApplication,"","en")).await;});
264 }
267impl Drop for OpenedRoute {
268 fn drop(&mut self) {
269 let handles = std::mem::take(&mut self.handles);
270 if !handles.is_empty() {
271 // No runtime exists during process shutdown; the OS then closes the sockets.
272 let Ok(runtime) = tokio::runtime::Handle::try_current() else { return };
273 runtime.spawn(async move {
274 for handle in handles.into_iter().rev() {
275 let _ = tokio::time::timeout(Duration::from_secs(2), handle.disconnect(russh::Disconnect::ByApplication, "", "en")).await;
276 }
277 });
278 }
279 }
282/// Internal probe snapshot. Never serialized or exposed over IPC. A Windows sshd
283/// restart may close the original session before its guarded login probe runs.
284#[derive(Clone)]
285pub struct ConnectionTarget {
286 connection_epoch:u64,
287 vault_generation:Option<u64>,
288 route: Vec<Value>,
289 jump_credentials: Vec<Credentials>,
290 address: String,
293impl ConnectionTarget {
294 /// The same verified server after a change of its address. The host key still has to match.
295 pub fn at_address(&mut self, address: String) {
296 self.address = address;
297 }
300pub struct SessionManager {
301 pub profiles: Arc<ProfileService>,
302 pub logger: Arc<Logger>,
303 emit: Emitter,
304 sessions: Mutex<HashMap<String, Arc<SessionRecord>>>,
305 pending_trust: Mutex<HashMap<String, PendingTrust>>,
306 connection_epoch: AtomicU64,
307 connection_cancel:watch::Sender<u64>,
308 #[cfg(test)]
309 before_registration: Mutex<Option<Arc<dyn Fn() + Send + Sync>>>,
310 /// Internal subscribers (Transfer Manager) receive every public state change.
311 pub state_events: broadcast::Sender<Value>,
314impl SessionManager {
315 pub fn new(profiles: Arc<ProfileService>, logger: Arc<Logger>, emit: Emitter) -> Arc<Self> {
316 let (state_events, _) = broadcast::channel(256);
317 let (connection_cancel,_)=watch::channel(0);
318 Arc::new(Self { profiles, logger, emit, sessions: Mutex::new(HashMap::new()), pending_trust: Mutex::new(HashMap::new()), connection_epoch: AtomicU64::new(0), connection_cancel,
319 #[cfg(test)] before_registration: Mutex::new(None), state_events })
320 }
322 fn emit_state(&self, status: Value) {
323 let _ = self.state_events.send(status.clone());
324 (self.emit)("ssh:state", status);
325 }
327 /// Re-publishes the state of a live session (latency updates).
328 pub fn publish_state(&self, record: &Arc<SessionRecord>) {
329 let current = self.sessions.lock().get(&record.session_id).map(|r| Arc::ptr_eq(r, record)).unwrap_or(false);
330 if current && !record.finalized.load(Ordering::SeqCst) {
331 self.emit_state(record.public_status());
332 }
333 }
335 pub fn list(&self) -> Vec<Value> {
336 self.sessions.lock().values().map(|r| r.public_status()).collect()
337 }
339 pub fn get(&self, session_id: &str) -> Option<Arc<SessionRecord>> {
340 self.sessions.lock().get(session_id).cloned()
341 }
343 pub fn require_connected(&self, session_id: &str) -> CoreResult<Arc<SessionRecord>> {
344 let record = self.get(session_id).ok_or_else(not_connected)?;
345 if !record.is_connected() || record.shell.lock().is_none() {
346 return Err(not_connected());
347 }
348 Ok(record)
349 }
351 fn credentials(&self, profile: &Value, options: &Value) -> CoreResult<Credentials> {
352 self.credentials_fields(profile, options.get("password"), options.get("passphrase"))
353 }
355 fn credentials_fields(&self, profile: &Value, password: Option<&Value>, passphrase: Option<&Value>) -> CoreResult<Credentials> {
356 if profile["authMethod"] == "privateKey" {
357 let id = profile["id"].as_str().unwrap_or("");
358 let saved = !id.is_empty() && !id.starts_with("adhoc:");
359 let keychain_key_id = if saved { profile["keychainKeyId"].as_str().unwrap_or("") } else { "" };
360 let text = Zeroizing::new(if keychain_key_id.is_empty() {
361 read_private_key_file(profile["privateKeyPath"].as_str().unwrap_or(""))?
362 } else {
363 self.profiles.keychain_private_key(keychain_key_id)?
364 });
365 // An explicit passphrase wins; otherwise a saved profile may hold the passphrase in the vault.
366 let passphrase = match passphrase.and_then(Value::as_str).filter(|p| !p.is_empty()) {
367 Some(explicit) => Some(Zeroizing::new(explicit.to_string())),
368 None if saved => self.profiles.get_stored_password_for_connection(profile)?.map(Zeroizing::new),
369 None => None,
370 };
371 let key = decode_private_key(&text, passphrase.as_deref().map(|s| s.as_str()))?;
372 Ok(Credentials::Key(Arc::new(key)))
373 } else {
374 let id = profile["id"].as_str().unwrap_or("");
375 let password = if let Some(explicit) = password.and_then(Value::as_str).filter(|p| !p.is_empty()) {
376 explicit.to_string()
377 } else if id.starts_with("adhoc:") { String::new() }
378 else { self.profiles.get_stored_password_for_connection(profile)?.unwrap_or_default() };
379 if password.is_empty() {
380 return Err(CoreError::new("PASSWORD_REQUIRED", "Enter the SSH password."));
381 }
382 Ok(Credentials::Password(Zeroizing::new(password)))
383 }
384 }
386 async fn resolve_address(&self, host: &str, port: u16) -> CoreResult<IpAddr> {
387 if let Ok(ip) = host.parse::<IpAddr>() {
388 return Ok(ip);
389 }
390 let addresses: Vec<SocketAddr> = match tokio::net::lookup_host((host, port)).await {
391 Ok(list) => list.collect(),
392 Err(_) => Vec::new(),
393 };
394 let mut ips: Vec<IpAddr> = addresses.iter().map(|a| a.ip()).collect();
395 let mut source = "system";
396 if ips.is_empty() {
397 ips = crate::core::diagnostics::dns::resolve_both(host).await;
398 source = "dns";
399 }
400 let selected = ips.iter().find(|ip| ip.is_ipv4()).or_else(|| ips.first()).copied().ok_or_else(|| {
401 CoreError::new("SSH_DNS_RESOLUTION_FAILED", format!("DNS name \"{host}\" could not be resolved.")).with("host", host)
402 })?;
403 self.logger.info(
404 "SSH hostname resolved.",
405 json!({ "addressFamily": if selected.is_ipv4() { 4 } else { 6 }, "addressCount": ips.len(), "source": source }),
406 );
407 Ok(selected)
408 }
410 fn normalize_error(&self, error: &russh::Error, details: &[(&str, Value)]) -> CoreError {
411 let message = error.to_string();
412 let mut code = "SSH_CONNECTION_FAILED";
413 if let russh::Error::IO(io) = error {
414 let raw = io.raw_os_error();
415 if cfg!(target_os = "macos") && matches!(raw, Some(1) | Some(13)) {
416 code = "SSH_LOCAL_NETWORK_DENIED";
417 } else if cfg!(target_os = "macos") && matches!(raw, Some(65) | Some(51)) {
418 code = "SSH_MAC_NETWORK_UNREACHABLE";
419 } else if io.kind() == std::io::ErrorKind::ConnectionRefused {
420 code = "SSH_CONNECTION_REFUSED";
421 } else if io.kind() == std::io::ErrorKind::TimedOut {
422 code = "SSH_TIMEOUT";
423 }
424 } else if matches!(error, russh::Error::ConnectionTimeout | russh::Error::KeepaliveTimeout | russh::Error::InactivityTimeout | russh::Error::Elapsed(_)) {
425 code = "SSH_TIMEOUT";
426 } else if matches!(error, russh::Error::NotAuthenticated | russh::Error::NoAuthMethod) {
427 code = "SSH_AUTHENTICATION_FAILED";
428 }
429 let mut result = CoreError::new(code, message);
430 for (key, value) in details {
431 result = result.with(key, value.clone());
432 }
433 result
434 }
436 fn manual_profile(&self, input: &Value) -> CoreResult<Value> {
437 if !input.is_object() || !matches!(input["authMethod"].as_str(), Some("password" | "privateKey")) {
438 return Err(CoreError::new("SSH_TARGET_INVALID", "Invalid manual SSH target."));
439 }
440 let mut profile = normalize_profile(input, None)
441 .map_err(|e| CoreError::new("SSH_TARGET_INVALID", e.message))?;
442 profile["id"] = json!(format!("adhoc:{}", uuid::Uuid::new_v4()));
443 profile["label"] = json!(format!("{}@{}", profile["username"].as_str().unwrap_or(""), profile["host"].as_str().unwrap_or("")));
444 profile["jumpProfileId"] = json!("");
445 Ok(profile)
446 }
448 fn route_profiles(&self, options: &Value) -> CoreResult<Vec<Value>> {
449 let profile = match (options.get("profileId"), options.get("adhoc")) {
450 (Some(Value::String(id)), None) if !id.is_empty() => self.profiles.get(id)?,
451 (None, Some(input)) => self.manual_profile(input)?,
452 _ => return Err(CoreError::new("SSH_TARGET_INVALID", "Choose exactly one saved or manual SSH target.")),
453 };
454 let mut seen = std::collections::HashSet::from([profile["id"].as_str().unwrap_or("").to_string()]);
455 let saved_jump = |id: &str| self.profiles.get(id).map_err(|_| CoreError::new("JUMP_PROFILE_NOT_FOUND", "Jump profile was not found.")
456 .with("hop", "jump").with("hopHost", "").with("jumpProfileId", id));
457 let mut next = match options.get("jump") {
458 Some(Value::Null) => None,
459 None => profile["jumpProfileId"].as_str().filter(|id| !id.is_empty()).map(saved_jump).transpose()?,
460 Some(jump) if jump.is_object() && jump.get("profileId").is_some() => {
461 if jump.get("host").is_some() {
462 return Err(CoreError::new("SSH_TARGET_INVALID", "Choose one jump host."));
463 }
464 Some(saved_jump(jump["profileId"].as_str().unwrap_or(""))?)
465 }
466 Some(jump) => Some(self.manual_profile(jump)?),
467 };
468 let mut jumps = Vec::new();
469 while let Some(hop) = next {
470 if !seen.insert(hop["id"].as_str().unwrap_or("").to_string()) {
471 return Err(self.hop_error(CoreError::new("JUMP_LOOP", "The jump route contains a cycle."), &hop, true));
472 }
473 if jumps.len() == 3 {
474 return Err(self.hop_error(CoreError::new("JUMP_TOO_DEEP", "At most three jump hosts are allowed."), &hop, true));
475 }
476 next = hop["jumpProfileId"].as_str().filter(|id| !id.is_empty()).map(saved_jump).transpose()?;
477 jumps.push(hop);
478 }
479 jumps.reverse();
480 jumps.push(profile);
481 Ok(jumps)
482 }
484 fn hop_error(&self, error: CoreError, profile: &Value, jump: bool) -> CoreError {
485 let mut error = error.with("host", profile["host"].clone()).with("port", profile["port"].clone())
486 .with("hop", if jump { "jump" } else { "target" });
487 if jump {
488 error = error.with("hopHost", profile["host"].clone());
489 }
490 error
491 }
493 fn route_credentials(&self, route: &[Value], options: &Value) -> CoreResult<Vec<Credentials>> {
494 route[..route.len() - 1].iter().enumerate().map(|(i, hop)| {
495 let (password, passphrase) = if i + 2 == route.len() {
496 (options.get("jumpPassword"), options.get("jumpPassphrase"))
497 } else { (None, None) };
498 self.credentials_fields(hop, password, passphrase).map_err(|e| self.hop_error(e, hop, true))
499 }).collect()
500 }
502 /// Shared handshake, host-key verification and authentication for terminals and
503 /// fresh logins. Only the first transport is resolved locally.
504 async fn open_route(&self, route: &[Value], jumps: &[Credentials], target: &Credentials, address: Option<&str>) -> CoreResult<OpenedRoute> {
505 self.profiles.assert_connection_route(route)?;
506 let mut opened = OpenedRoute::default();
507 for (i, profile) in route.iter().enumerate() {
508 let jump = i + 1 < route.len();
509 let host = profile["host"].as_str().unwrap_or("");
510 let port = profile["port"].as_u64().unwrap_or(22) as u16;
511 let known = self.profiles.get_host_key(host, port).map_err(|e| self.hop_error(e, profile, jump))?;
512 let observed = Arc::new(Mutex::new(None));
513 let server_id = Arc::new(Mutex::new(String::new()));
514 opened.server_id = server_id.clone();
515 let creds = if jump { &jumps[i] } else { target };
516 let authentication=match creds{Credentials::Key(key)=>key.algorithm().as_str().to_string(),Credentials::Password(_)=>"password".into()};
517 let negotiated=Arc::new(Mutex::new(json!({"exchanges":0,"authenticationAlgorithm":authentication,"authenticationPostQuantum":authentication=="ssh-mldsa44-ed25519"})));
518 opened.negotiated.push(negotiated.clone());
519 let policy=profile["kexPolicy"].as_str().unwrap_or("prefer-hybrid");
520 if !["prefer-hybrid","hybrid-only"].contains(&policy){return Err(CoreError::new("SSH_KEX_POLICY_INVALID","Unsupported SSH key exchange policy."));}
521 let handler = ClientHandler { expected: known.clone(), observed: observed.clone(), server_id: server_id.clone(),negotiated };
522 let attempt = tokio::time::timeout(READY_TIMEOUT, async {
523 let handle = if let Some(previous) = opened.handles.last() {
524 let channel = previous.channel_open_direct_tcpip(host, u32::from(port), "127.0.0.1", 0).await
525 .map_err(|_| ConnectFailure::Forward)?;
526 client::connect_stream(client_config_policy(true,policy), channel.into_stream(), handler).await.map_err(ConnectFailure::Ssh)?
527 } else {
528 let ip = match address {
529 Some(address) => address.parse::<IpAddr>().map_err(|_| ConnectFailure::Core(CoreError::new("SSH_CONNECTION_FAILED", "Invalid first-hop address.")))?,
530 None => self.resolve_address(host, port).await.map_err(ConnectFailure::Core)?,
531 };
532 opened.address = ip.to_string();
533 client::connect(client_config_policy(true,policy), SocketAddr::new(ip, port), handler).await.map_err(ConnectFailure::Ssh)?
534 };
535 let mut pending=PendingHandle(Some(handle));let handle=pending.0.as_mut().unwrap();
536 self.profiles.assert_connection_route(route).map_err(ConnectFailure::Core)?;
537 let result = match creds {
538 Credentials::Password(password) => handle.authenticate_password(profile["username"].as_str().unwrap_or(""), password.as_str()).await,
539 Credentials::Key(key) => {
540 let hash = handle.best_supported_rsa_hash().await.map_err(ConnectFailure::Ssh)?.flatten();
541 handle.authenticate_publickey(profile["username"].as_str().unwrap_or(""), PrivateKeyWithHashAlg::new(key.clone(), hash)).await
542 }
543 }.map_err(ConnectFailure::Ssh)?;
544 if !result.success() {
545 let _ = handle.disconnect(russh::Disconnect::ByApplication, "", "en").await;
546 return Err(ConnectFailure::Auth);
547 }
548 Ok::<_, ConnectFailure>(pending.0.take().unwrap())
549 }).await;
550 match attempt {
551 Ok(Ok(handle)) => opened.handles.push(Arc::new(handle)),
552 failure => {
553 let fingerprint = observed.lock().clone();
554 let error = match failure {
555 Ok(Err(ConnectFailure::Ssh(russh::Error::UnknownKey))) if fingerprint.is_some() => {
556 let mut error = CoreError::new(if known.is_some() { "HOST_KEY_MISMATCH" } else { "HOST_KEY_UNKNOWN" }, "SSH host key verification requires confirmation.")
557 .with("fingerprint", fingerprint.unwrap());
558 if let Some(known) = known { error = error.with("knownFingerprint", known); }
559 error
560 }
561 Ok(Err(ConnectFailure::Forward)) => {
562 // The refusal belongs to the forwarding jump, not the next host.
563 return Err(self.hop_error(CoreError::new("JUMP_CHANNEL_FAILED", "The jump host refused forwarding or could not reach the next host.").with("nextHost", host).with("nextPort", port), &route[i - 1], true));
564 }
565 Ok(Err(ConnectFailure::Ssh(e))) => self.normalize_error(&e, &[]),
566 Ok(Err(ConnectFailure::Auth)) => CoreError::new("SSH_AUTHENTICATION_FAILED", "All configured authentication methods failed."),
567 Ok(Err(ConnectFailure::Core(e))) => e,
568 Err(_) => CoreError::new("SSH_TIMEOUT", "Timed out while waiting for handshake."),
569 Ok(Ok(_)) => unreachable!(),
570 };
571 return Err(self.hop_error(error, profile, jump));
572 }
573 }
574 }
575 Ok(opened)
576 }
578 pub async fn connect(self: &Arc<Self>, options: &Value) -> CoreResult<Value> {
579 let connection_epoch = self.connection_epoch.load(Ordering::SeqCst);
580 let session_id = options.get("sessionId").and_then(Value::as_str).unwrap_or("").trim().to_string();
581 if !SESSION_ID.is_match(&session_id) {
582 return Err(CoreError::new("INVALID_SESSION_ID", "Invalid terminal session ID."));
583 }
584 let slot = match options.get("slot") {
585 None | Some(Value::Null) => None,
586 Some(v) if v.as_u64() == Some(1) || v.as_u64() == Some(2) => Some(v.as_u64().unwrap() as u8),
587 _ => return Err(CoreError::new("SSH_TARGET_INVALID", "Invalid terminal slot.")),
588 };
589 let route = self.route_profiles(options)?;
590 let profile = route.last().unwrap().clone();
591 let jumps = self.route_credentials(&route, options)?;
592 let credentials = self.credentials(&profile, options).map_err(|e| self.hop_error(e, &profile, false))?;
593 let (lost_tx, _) = watch::channel(false);
594 let vault_generation=if route.iter().any(|p|!p["id"].as_str().unwrap_or("").starts_with("adhoc:")){Some(self.profiles.vault.unlocked_generation()?)}else{None};
595 let record = Arc::new(SessionRecord {
596 connection_epoch,vault_generation,
597 session_id: session_id.clone(),
598 profile_id: profile["id"].as_str().unwrap_or("").to_string(),
599 label: profile["label"].as_str().unwrap_or("").to_string(),
600 host: profile["host"].as_str().unwrap_or("").to_string(),
601 address: Mutex::new(profile["host"].as_str().unwrap_or("").to_string()),
602 port: profile["port"].as_u64().unwrap_or(22) as u16,
603 username: profile["username"].as_str().unwrap_or("").to_string(),
604 auth_method: profile["authMethod"].as_str().unwrap_or("password").to_string(),
605 slot,
606 adhoc: profile["id"].as_str().unwrap_or("").starts_with("adhoc:"),
607 profile,
608 route,
609 jump_credentials: Mutex::new(None),
610 transport: Mutex::new(None),
611 state: Mutex::new("connecting".into()),
612 handle: Mutex::new(None),
613 shell: Mutex::new(None),
614 latency: Mutex::new(None),
615 server_id: Mutex::new(String::new()),
616 finalized: AtomicBool::new(false),
617 lifecycle: Mutex::new(()),
618 lost_tx,
619 platform: Mutex::new(None),
620 privilege: Mutex::new(None),
621 sudo_cache: Mutex::new(None),
622 windows_queue: tokio::sync::Mutex::new(()),
623 metrics_state: Mutex::new(json!({})),
624 stop_latency: Mutex::new(None),
625 });
626 #[cfg(test)] {
627 let hook = self.before_registration.lock().take();
628 if let Some(hook) = hook { hook(); }
629 }
630 let previous = {
631 let mut sessions = self.sessions.lock();
632 if self.connection_epoch.load(Ordering::SeqCst) != connection_epoch { return Err(not_connected()); }
633 self.pending_trust.lock().remove(&session_id);
634 sessions.insert(session_id.clone(), record.clone())
635 };
636 if let Some(previous) = previous { self.finalize(&previous, "reconnect"); }
637 self.emit_state(record.public_status());
638 let cols = normalize_terminal_size(options.get("cols"), 100, 20, 500);
639 let rows = normalize_terminal_size(options.get("rows"), 30, 5, 300);
640 let mut lost = record.lost();
641 let result = {
642 let attempt = async {
643 let transport = self.open_route(&record.route, &jumps, &credentials, None).await?;
644 let handle = transport.handles.last().unwrap().clone();
645 let channel = tokio::time::timeout(READY_TIMEOUT, async {
646 let channel = handle.channel_open_session().await?;
647 channel.request_pty(false, "xterm-256color", cols, rows, 0, 0, &[]).await?;
648 channel.request_shell(true).await?;
649 Ok::<_, russh::Error>(channel)
650 }).await.map_err(|_| CoreError::new("SSH_TIMEOUT", "Timed out opening the terminal."))?
651 .map_err(|e| self.normalize_error(&e, &[]))?;
652 Ok::<_, CoreError>((transport, handle, channel))
653 };
654 let cancelled = *lost.borrow();
655 if cancelled { Err(not_connected()) } else {
656 tokio::select! {
657 result = attempt => result,
658 _ = lost.changed() => Err(not_connected()),
659 }
660 }
661 };
662 match result {
663 Ok((transport, handle, channel)) => {
664 let status = {
665 let _lifecycle = record.lifecycle.lock();
666 let sessions = self.sessions.lock();
667 let current = sessions.get(&session_id).is_some_and(|r| Arc::ptr_eq(r, &record));
668 if !current || record.finalized.load(Ordering::SeqCst) { return Err(not_connected()); }
669 self.pending_trust.lock().remove(&session_id);
670 drop(sessions);
671 if !record.has_jump() { *record.address.lock() = transport.address.clone(); }
672 *record.handle.lock() = Some(handle);
673 *record.server_id.lock() = transport.server_id.lock().clone();
674 *record.transport.lock() = Some(transport);
675 *record.jump_credentials.lock() = Some(jumps);
676 *record.state.lock() = "connected".into();
677 self.bind_shell(&record, channel);
678 if !record.adhoc { let _ = self.profiles.mark_used(&record.profile_id); }
679 record.public_status()
680 };
681 self.logger.info("SSH session connected.", json!({ "authMethod": record.auth_method }));
682 self.publish_state(&record);
683 crate::core::ssh::latency::start(self.clone(), record.clone());
684 let manager = self.clone();
685 tokio::spawn(async move {
686 let mut lost = record.lost();
687 loop {
688 tokio::select! {
689 _ = lost.changed() => break,
690 _ = tokio::time::sleep(Duration::from_millis(250)) => {
691 let closed = record.transport.lock().as_ref().map(|t| t.handles.iter().any(|h| h.is_closed())).unwrap_or(true);
692 if closed { manager.finalize(&record, "route-close"); break; }
693 }
694 }
695 }
696 });
697 Ok(status)
698 }
699 Err(error) => {
700 let error = error.with("sessionId", session_id.clone()).with("profileId", if record.adhoc { "" } else { &record.profile_id });
701 let lifecycle = record.lifecycle.lock();
702 let sessions = self.sessions.lock();
703 let current = sessions.get(&record.session_id).is_some_and(|r| Arc::ptr_eq(r, &record));
704 if current && !record.finalized.load(Ordering::SeqCst) && matches!(error.code.as_str(), "HOST_KEY_UNKNOWN" | "HOST_KEY_MISMATCH") {
705 self.pending_trust.lock().insert(session_id, PendingTrust {
706 profile_id: if record.adhoc { String::new() } else { record.profile_id.clone() },
707 host: error.details["host"].as_str().unwrap_or("").to_string(),
708 port: error.details["port"].as_u64().unwrap_or(22) as u16,
709 fingerprint: error.details["fingerprint"].as_str().unwrap_or("").to_string(),
710 known: error.details.get("knownFingerprint").and_then(Value::as_str).map(str::to_string),
711 expires: Instant::now() + PENDING_TRUST_TTL,
712 });
713 }
714 drop(sessions);
715 drop(lifecycle);
716 self.logger.warn("SSH connection failed.", json!({ "code": error.code }));
717 self.finalize(&record, &error.code);
718 Err(error)
719 }
720 }
721 }
723 fn bind_shell(self: &Arc<Self>, record: &Arc<SessionRecord>, channel: russh::Channel<Msg>) {
724 let (tx, mut rx) = mpsc::unbounded_channel::<ShellCommand>();
725 *record.shell.lock() = Some(tx);
726 let (mut read, write): (_, ChannelWriteHalf<Msg>) = channel.split();
727 let manager = self.clone();
728 let record = record.clone();
729 tokio::spawn(async move {
730 let mut decoder = Utf8Decoder::default();
731 let mut lost = record.lost();
732 let reason = loop {
733 if !record.is_connected() { break "disconnected"; }
734 tokio::select! {
735 message = read.wait() => match message {
736 Some(ChannelMsg::Data { data }) => {
737 let text = decoder.write(&data);
738 if !text.is_empty() && record.is_connected() { (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": text })); }
739 }
740 Some(ChannelMsg::ExtendedData { data, .. }) => {
741 let text = decoder.write(&data);
742 if !text.is_empty() && record.is_connected() { (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": text })); }
743 }
744 Some(ChannelMsg::Close) | None => break "shell-close",
745 Some(ChannelMsg::Failure) => {
746 (manager.emit)("ssh:error", json!({ "sessionId": record.session_id, "message": "The SSH server refused the interactive shell." }));
747 break "shell-error";
748 }
749 Some(_) => {}
750 },
751 command = rx.recv() => match command {
752 Some(ShellCommand::Input(bytes)) => { if write.data_bytes(bytes).await.is_err() { break "shell-error"; } }
753 Some(ShellCommand::Resize(cols, rows)) => { let _ = write.window_change(cols, rows, 0, 0).await; }
754 None => break "shell-close",
755 },
756 _ = lost.changed() => break "disconnected",
757 }
758 };
759 let trailing = decoder.end();
760 if !trailing.is_empty() && record.is_connected() {
761 (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": trailing }));
762 }
763 let _ = write.close().await;
764 manager.finalize(&record, reason);
765 });
766 }
768 pub fn trust_pending(&self, options: &Value) -> CoreResult<Value> {
769 let session_id = options.get("sessionId").and_then(Value::as_str).unwrap_or("").to_string();
770 let fingerprint = options.get("fingerprint").and_then(Value::as_str).unwrap_or("").to_lowercase();
771 let pending = self.pending_trust.lock().get(&session_id).cloned();
772 let Some(pending) = pending.filter(|p| p.expires >= Instant::now()) else {
773 self.pending_trust.lock().remove(&session_id);
774 return Err(CoreError::new("HOST_KEY_CONFIRMATION_EXPIRED", "The host-key confirmation expired."));
775 };
776 if options.get("profileId").and_then(Value::as_str) != Some(pending.profile_id.as_str()) || pending.fingerprint != fingerprint {
777 return Err(CoreError::new("HOST_KEY_CONFIRMATION_INVALID", "The host-key confirmation does not match the pending connection."));
778 }
779 let replace = options.get("replaceKnown") == Some(&Value::Bool(true));
780 if let Some(known) = &pending.known {
781 let confirmed = options.get("knownFingerprint").and_then(Value::as_str).unwrap_or("").to_lowercase();
782 if !replace || &confirmed != known {
783 return Err(CoreError::new(
784 "HOST_KEY_REPLACEMENT_CONFIRMATION_REQUIRED",
785 "Replacing a changed SSH host key requires explicit confirmation of both fingerprints.",
786 ));
787 }
788 if self.profiles.get_host_key(&pending.host, pending.port)?.as_deref() != Some(known.as_str()) {
789 self.pending_trust.lock().remove(&session_id);
790 return Err(CoreError::new("HOST_KEY_REPLACEMENT_STALE", "The stored SSH host key changed while confirmation was pending."));
791 }
792 } else if replace {
793 return Err(CoreError::new("HOST_KEY_CONFIRMATION_INVALID", "A new host key cannot be confirmed as a replacement."));
794 }
795 let result = self.profiles.confirm_host_key(&pending.host, pending.port, &pending.fingerprint, pending.known.as_deref())?;
796 self.pending_trust.lock().remove(&session_id);
797 let details = json!({});
798 if pending.known.is_some() {
799 self.logger.warn("Changed SSH host key replaced by user confirmation.", details);
800 } else {
801 self.logger.info("SSH host key trusted by user.", details);
802 }
803 Ok(result)
804 }
806 /// Terminal input. Called synchronously and in order; data is queued to the shell task.
807 pub fn write(&self, session_id: &str, data: &str) -> CoreResult<()> {
808 let record = self.require_connected(session_id)?;
809 if data.len() > MAX_INPUT_LENGTH {
810 return Err(CoreError::new("TERMINAL_INPUT_TOO_LARGE", "Terminal input is too large."));
811 }
812 let sender = record.shell.lock().clone().ok_or_else(not_connected)?;
813 sender.send(ShellCommand::Input(data.as_bytes().to_vec())).map_err(|_| not_connected())
814 }
816 pub fn resize(&self, session_id: &str, cols: Option<&Value>, rows: Option<&Value>) -> CoreResult<Value> {
817 let record = self.require_connected(session_id)?;
818 let cols = normalize_terminal_size(cols, 100, 20, 500);
819 let rows = normalize_terminal_size(rows, 30, 5, 300);
820 if let Some(sender) = record.shell.lock().clone() {
821 let _ = sender.send(ShellCommand::Resize(cols, rows));
822 }
823 Ok(json!({ "cols": cols, "rows": rows }))
824 }
826 /// The fingerprint this session's server is trusted with now.
827 pub fn trusted_host_key(&self, session_id: &str) -> Option<String> {
828 let record = self.get(session_id)?;
829 self.profiles.get_host_key(&record.host, record.port).ok().flatten()
830 }
832 /// Availability of a jump-only server, through a session already open to its jump host:
833 /// one forwarding channel, no login and no new session. Never an unproven "online".
834 pub async fn probe_through_jump(&self, jump_profile_id: &str, host: &str, port: u16, timeout: Duration) -> Outcome {
835 let handle = self.sessions.lock().values()
836 .find(|record| record.profile_id == jump_profile_id && record.state.lock().as_str() == "connected")
837 .and_then(|record| record.handle.lock().clone());
838 let Some(handle) = handle else { return Outcome::Unverified { reason: "noJumpSession" } };
839 let start = std::time::Instant::now();
840 match tokio::time::timeout(timeout, handle.channel_open_direct_tcpip(host, u32::from(port), "127.0.0.1", 0)).await {
841 Ok(Ok(channel)) => {
842 let ms = (start.elapsed().as_secs_f64() * 10_000.0).round() / 10.0;
843 let _ = channel.close().await;
844 Outcome::Online { ms }
845 }
846 Ok(Err(russh::Error::ChannelOpenFailure(russh::ChannelOpenFailure::ConnectFailed))) => Outcome::Failed { reason: "refused" },
847 Ok(Err(_)) => Outcome::Unverified { reason: "jumpDenied" },
848 Err(_) => Outcome::Failed { reason: "timeout" },
849 }
850 }
852 pub async fn disconnect(&self, session_id: &str, reason: &str) -> Value {
853 self.pending_trust.lock().remove(session_id);
854 let record = self.sessions.lock().get(session_id).cloned();
855 match record {
856 Some(record) => {
857 self.finalize(&record, reason);
858 json!({ "disconnected": true })
859 }
860 None => json!({ "disconnected": false }),
861 }
862 }
864 pub async fn disconnect_all(&self, reason: &str) -> Value {
865 let ids: Vec<String> = {
866 let sessions = self.sessions.lock();
867 let epoch=self.connection_epoch.fetch_add(1, Ordering::SeqCst)+1;
868 self.connection_cancel.send_replace(epoch);
869 sessions.keys().cloned().collect()
870 };
871 for id in &ids {
872 self.disconnect(id, reason).await;
873 }
874 self.pending_trust.lock().clear();
875 json!({ "disconnected": ids.len() })
876 }
878 pub fn finalize(&self, record: &Arc<SessionRecord>, reason: &str) {
879 let lifecycle = record.lifecycle.lock();
880 if record.finalized.swap(true, Ordering::SeqCst) {
881 return;
882 }
883 *record.state.lock() = "disconnected".into();
884 if let Some(task) = record.stop_latency.lock().take() {
885 task.abort();
886 }
887 *record.latency.lock() = None;
888 record.lost_tx.send_replace(true);
889 *record.sudo_cache.lock() = None;
890 {
891 let mut sessions = self.sessions.lock();
892 if sessions.get(&record.session_id).map(|r| Arc::ptr_eq(r, record)).unwrap_or(false) {
893 sessions.remove(&record.session_id);
894 }
895 }
896 record.shell.lock().take();
897 record.handle.lock().take();
898 record.transport.lock().take();
899 record.jump_credentials.lock().take();
900 let mut status = record.public_status();
901 status["reason"] = Value::String(reason.to_string());
902 drop(lifecycle);
903 self.emit_state(status);
904 self.logger.info("SSH session disconnected.", json!({}));
905 }
907 // -----------------------------------------------------------------------------------------
908 // Exec channels
909 // -----------------------------------------------------------------------------------------
911 pub async fn exec_channel(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration, limit: usize) -> CoreResult<ExecChannel> {
912 let handle = record.handle()?;
913 // A longer command makes dropbear drop the whole connection, including the terminal.
914 if command_too_long_for(&record.server_id.lock(), command) {
915 return Err(CoreError::new("REMOTE_COMMAND_TOO_LONG", "This SSH server (dropbear) does not accept commands this long, so this function is not available on it."));
916 }
917 let lost = record.lost();
918 let opened = tokio::time::timeout(timeout, async {
919 let channel = handle.channel_open_session().await?;
920 channel.exec(true, command.as_bytes().to_vec()).await?;
921 Ok::<_, russh::Error>(channel)
922 })
923 .await
924 .map_err(|_| CoreError::new("REMOTE_COMMAND_TIMEOUT", "The remote command timed out."))?
925 .map_err(|e| if record.is_connected() { CoreError::new("REMOTE_COMMAND_FAILED", e.to_string()) } else { not_connected() })?;
926 let (read, write) = opened.split();
927 Ok(ExecChannel::new(read, write, timeout, limit, lost))
928 }
930 /// Runs a fixed command and requires exit code 0 (`_execFixed`).
931 pub async fn exec_fixed(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration, limit: usize) -> CoreResult<ExecResult> {
932 let mut channel = self.exec_channel(record, command, timeout, limit).await?;
933 channel.eof().await;
934 let result = channel.finish().await?;
935 match result.code {
936 Some(0) => Ok(result),
937 code => {
938 let message = if result.stderr.trim().is_empty() { format!("Remote command exited with {}.", code.map(|c| c.to_string()).unwrap_or_else(|| "undefined".into())) } else { result.stderr.trim().to_string() };
939 Err(CoreError::new("REMOTE_COMMAND_FAILED", message).with("remoteExitCode", code.map(Value::from).unwrap_or(Value::Null)))
940 }
941 }
942 }
944 /// Runs a matrix operation over an exec channel without privilege handling. The arguments are quoted.
945 pub async fn exec_operation(&self, record: &Arc<SessionRecord>, operation: &str, args: &[String], timeout: Duration, limit: usize) -> CoreResult<ExecResult> {
946 let platform = crate::core::ssh::platform::detect_platform(self, record, false).await?;
947 let binding = crate::core::matrix::binding(operation, crate::core::matrix::os_of(&platform))?;
948 self.exec_fixed(record, &crate::core::matrix::command(binding.script, args), timeout, limit).await
949 }
951 pub async fn exec_fixed_default(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration) -> CoreResult<ExecResult> {
952 self.exec_fixed(record, command, timeout, MAX_COMMAND_OUTPUT).await
953 }
955 /// One transport round trip (keepalive@openssh.com with want-reply), fallback `exit 0`.
956 pub async fn ssh_round_trip(&self, record: &Arc<SessionRecord>) -> CoreResult<(Option<f64>, &'static str)> {
957 let handle = record.handle()?;
958 let started = Instant::now();
959 match tokio::time::timeout(Duration::from_secs(5), handle.send_ping()).await {
960 Ok(Ok(())) => return Ok((Some(started.elapsed().as_secs_f64() * 1000.0), "transport")),
961 Ok(Err(_)) => {}
962 Err(_) => return Err(CoreError::new("SSH_RTT_TIMEOUT", "The SSH server did not answer the round-trip request in time.")),
963 }
964 let started = Instant::now();
965 self.exec_fixed_default(record, crate::core::matrix::binding("ssh.latency.round-trip.read", "debian")?.script, Duration::from_secs(5)).await?;
966 Ok((Some(started.elapsed().as_secs_f64() * 1000.0), "command"))
967 }
969 /// Serializes PowerShell helpers of one Windows session (Win32-OpenSSH stalls on concurrent stdin helpers).
970 pub async fn windows_serial<T, F: std::future::Future<Output = CoreResult<T>>>(&self, record: &Arc<SessionRecord>, task: F) -> CoreResult<T> {
971 let _turn = record.windows_queue.lock().await;
972 task.await
973 }
975 // -----------------------------------------------------------------------------------------
976 // Identity facts, fresh logins (Users lockout protection)
977 // -----------------------------------------------------------------------------------------
979 pub fn session_auth_info(&self, session_id: &str) -> CoreResult<Value> {
980 let record = self.require_connected(session_id)?;
981 let fingerprint = if record.auth_method == "privateKey" {
982 match record.profile["keychainKeyId"].as_str().filter(|id| !id.is_empty()) {
983 Some(key_id) => self.profiles.keychain_public_fingerprint(key_id),
984 None => crate::core::ssh::keys::public_key_fingerprint_for_private_key(record.profile["privateKeyPath"].as_str().unwrap_or("")),
985 }.map(Value::String).unwrap_or(Value::Null)
986 } else { Value::Null };
987 // The target address is not the SSH client's source address. Users policy
988 // obtains the actual source from the remote SSH_CONNECTION environment.
989 Ok(json!({ "username": record.username, "authMethod": record.auth_method, "address": record.address.lock().clone(), "publicKeyFingerprint": fingerprint }))
990 }
992 pub fn probe_login_requirements(&self, session_id: &str, credentials: &Value) -> CoreResult<Value> {
993 let record = self.require_connected(session_id)?;
994 Ok(match self.credentials(&record.profile, credentials) {
995 Ok(_) => json!({ "ready": true, "missing": null }),
996 Err(e) => json!({ "ready": false, "missing": if e.code == "PASSWORD_REQUIRED" { "password" } else if matches!(e.code.as_str(), "PRIVATE_KEY_PASSPHRASE_REQUIRED" | "PRIVATE_KEY_PASSPHRASE_INVALID") { "passphrase" } else { "privateKey" } }),
997 })
998 }
1000 pub fn connection_target(&self, session_id: &str) -> CoreResult<ConnectionTarget> {
1001 let record = self.require_connected(session_id)?;
1002 let _lifecycle = record.lifecycle.lock();
1003 if !record.is_connected() { return Err(not_connected()); }
1004 let jump_credentials = record.jump_credentials.lock().clone().ok_or_else(not_connected)?;
1005 let address = record.address.lock().clone();
1006 let target=ConnectionTarget {
1007 connection_epoch:record.connection_epoch,vault_generation:record.vault_generation,
1008 route: record.route.clone(),
1009 jump_credentials,
1010 address,
1011 };
1012 self.assert_fresh_target(&target)?;Ok(target)
1015 /// A small final decision serialized with global disconnect and vault lock.
1016 /// The callback must not reenter SessionManager or VaultStore.
1017 fn with_fresh_target<R>(&self,target:&ConnectionTarget,operation:impl FnOnce()->CoreResult<R>)->CoreResult<R>{
1018 let _sessions=self.sessions.lock();
1019 if self.connection_epoch.load(Ordering::SeqCst)!=target.connection_epoch{return Err(CoreError::new("SSH_CONNECTION_CANCELLED","The SSH authorization was cancelled. Reconnect before starting a new login."));}
1020 if let Some(generation)=target.vault_generation{self.profiles.vault.with_generation(generation,operation)}else{operation()}
1022 fn assert_fresh_target(&self,target:&ConnectionTarget)->CoreResult<()> {self.with_fresh_target(target,||Ok(()))}
1024 /// One fixed command over a new connection following the original verified route.
1025 pub async fn fresh_exec(&self, target: &ConnectionTarget, credentials: &Value, command: &str, input: &str, timeout: Duration, accept_any_exit: bool) -> CoreResult<Value> {
1026 let probe_failed = |error: CoreError| {
1027 let mut wrapped = CoreError::new("SSH_PROBE_FAILED", format!("A new SSH login was not confirmed: {}", error.message));
1028 wrapped.details = error.details;
1029 wrapped.with("causeCode", error.code)
1030 };
1031 let mut cancelled=self.connection_cancel.subscribe();
1032 self.assert_fresh_target(target).map_err(probe_failed)?;
1033 let run = async {
1034 self.assert_fresh_target(target)?;
1035 self.profiles.assert_connection_route(&target.route)?;
1036 let profile = target.route.last().ok_or_else(not_connected)?;
1037 let creds = self.credentials(profile, credentials)?;
1038 let address = (target.route.len() == 1).then_some(target.address.as_str());
1039 let transport = self.open_route(&target.route, &target.jump_credentials, &creds, address).await?;
1040 self.assert_fresh_target(target)?;
1041 self.profiles.assert_connection_route(&target.route)?;
1042 let handle = transport.handles.last().unwrap();
1043 let channel = handle.channel_open_session().await.map_err(|e| self.normalize_error(&e, &[]))?;
1044 self.assert_fresh_target(target)?;
1045 channel.exec(true, command.as_bytes().to_vec()).await.map_err(|e| self.normalize_error(&e, &[]))?;
1046 let (read, write) = channel.split();
1047 let (_tx, lost) = watch::channel(false);
1048 let mut exec = ExecChannel::new(read, write, timeout, MAX_COMMAND_OUTPUT, lost);
1049 self.assert_fresh_target(target)?;
1050 if !input.is_empty() { exec.write(input.as_bytes()).await; }
1051 exec.eof().await;
1052 let result = exec.finish().await?;
1053 if !accept_any_exit && !matches!(result.code, Some(0) | None) {
1054 return Err(CoreError::new("SSH_PROBE_EXIT", format!("Probe command exited with {}.", result.code.unwrap_or(-1))));
1056 self.with_fresh_target(target,||Ok(json!({ "ok": true, "code": result.code, "stdout": result.stdout, "stderr": result.stderr, "address": target.address, "checkedAt": chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true) })))
1057 };
1058 tokio::select! {
1059 biased;
1060 _=cancelled.changed()=>Err(CoreError::new("SSH_CONNECTION_CANCELLED","The new SSH login was cancelled.")).map_err(probe_failed),
1061 result=tokio::time::timeout(timeout,run)=>result.map_err(|_|CoreError::new("SSH_TIMEOUT","Login probe timed out.")).and_then(|r|r).map_err(probe_failed),
1065 pub async fn probe_login(&self, session_id: &str, credentials: &Value) -> CoreResult<Value> {
1066 let target = self.connection_target(session_id)?;
1067 let result = self.fresh_exec(&target, credentials, crate::core::matrix::binding("ssh.login.probe", "debian")?.script, "", Duration::from_secs(60), true).await?;
1068 Ok(json!({ "ok": true, "address": result["address"], "checkedAt": result["checkedAt"] }))
1072enum ConnectFailure {
1073 Ssh(russh::Error),
1074 Auth,
1075 Forward,
1076 Core(CoreError),
1079#[cfg(test)]
1080mod tests {
1081 use super::*;
1083 #[test]
1084 fn terminal_size_bounds() {
1085 assert_eq!(normalize_terminal_size(Some(&json!(10)), 100, 20, 500), 20);
1086 assert_eq!(normalize_terminal_size(Some(&json!(9999)), 100, 20, 500), 500);
1087 assert_eq!(normalize_terminal_size(Some(&json!(80.5)), 100, 20, 500), 100);
1088 assert_eq!(normalize_terminal_size(None, 30, 5, 300), 30);
1091 #[test]
1092 fn quoting() {
1093 assert_eq!(shell_quote("a'b"), "'a'\"'\"'b'");
1097#[cfg(test)]
1098#[path = "../../../tests/support/ssh_route_unit.rs"]
1099mod route_tests;
SHA-2563471b9bbc5f7aac3dd094361e58128e858afa4a2a36f49514ad0113d5366f194

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

压缩包 SHA-256ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f
文件