CH-J Toolsمدیریت سرور از طریق SSH
منو
منبع منتشرشده

CH-J Tools

پوشه‌ها و فایل‌های یک نسخهٔ مشخص از برنامه را مرور کنید.

دانلود ZIP منبع
CH-J / Server Manager 872 فایل
core-20261030T220507Z-61e4558d889f افزوده‌شده: · ساخته‌شده:
session.rs افزوده‌شده: تاریخ ثبت نشده است ۴۷٫۴ KB · 969 خط
کد دانلود فایل
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, 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 }
42pub fn shell_quote(value: &str) -> String {
43 format!("'{}'", value.replace('\'', "'\"'\"'"))
46// ---------------------------------------------------------------------------------------------
47// Client handler: host-key verification
48// ---------------------------------------------------------------------------------------------
50pub struct ClientHandler {
51 expected: Option<String>,
52 observed: Arc<Mutex<Option<String>>>,
55impl client::Handler for ClientHandler {
56 type Error = russh::Error;
58 async fn check_server_key(&mut self, key: &PublicKeyOrCertificate) -> Result<bool, Self::Error> {
59 let blob = match key {
60 PublicKeyOrCertificate::PublicKey { key, .. } => key.to_bytes()?,
61 PublicKeyOrCertificate::Certificate(certificate) => certificate.to_bytes()?,
62 };
63 let fingerprint = sha256_hex(&blob);
64 *self.observed.lock() = Some(fingerprint.clone());
65 Ok(self.expected.as_deref() == Some(fingerprint.as_str()))
66 }
69/// Algorithm lists of ssh2 1.17 defaults used by 0.1.0 (plus ML-KEM hybrid KEX first).
70fn client_config(keepalive: bool) -> Arc<client::Config> {
71 use russh::{cipher, kex, mac, Preferred};
72 use std::borrow::Cow;
73 let preferred = Preferred {
74 kex: Cow::Owned(vec![
75 kex::MLKEM768X25519_SHA256,
76 kex::CURVE25519,
77 kex::CURVE25519_PRE_RFC_8731,
78 kex::ECDH_SHA2_NISTP256,
79 kex::ECDH_SHA2_NISTP384,
80 kex::ECDH_SHA2_NISTP521,
81 kex::DH_GEX_SHA256,
82 kex::DH_G14_SHA256,
83 kex::DH_G15_SHA512,
84 kex::DH_G16_SHA512,
85 kex::DH_G17_SHA512,
86 kex::DH_G18_SHA512,
87 kex::EXTENSION_SUPPORT_AS_CLIENT,
88 kex::EXTENSION_OPENSSH_STRICT_KEX_AS_CLIENT,
89 ]),
90 cipher: Cow::Owned(vec![
91 cipher::CHACHA20_POLY1305,
92 cipher::AES_128_GCM,
93 cipher::AES_256_GCM,
94 cipher::AES_128_CTR,
95 cipher::AES_192_CTR,
96 cipher::AES_256_CTR,
97 ]),
98 mac: Cow::Owned(vec![
99 mac::HMAC_SHA256_ETM,
100 mac::HMAC_SHA512_ETM,
101 mac::HMAC_SHA1_ETM,
102 mac::HMAC_SHA256,
103 mac::HMAC_SHA512,
104 mac::HMAC_SHA1,
105 ]),
106 ..Preferred::DEFAULT
107 };
108 Arc::new(client::Config {
109 preferred,
110 inactivity_timeout: None,
111 keepalive_interval: keepalive.then(|| Duration::from_secs(15)),
112 keepalive_max: 3,
113 nodelay: true,
114 ..Default::default()
115 })
118// ---------------------------------------------------------------------------------------------
119// Session records
120// ---------------------------------------------------------------------------------------------
122enum ShellCommand {
123 Input(Vec<u8>),
124 Resize(u32, u32),
127pub struct SessionRecord {
128 pub session_id: String,
129 pub profile_id: String,
130 pub label: String,
131 pub host: String,
132 pub address: Mutex<String>,
133 pub port: u16,
134 pub username: String,
135 pub auth_method: String,
136 pub slot: Option<u8>,
137 pub adhoc: bool,
138 profile: Value,
139 route: Vec<Value>,
140 jump_credentials: Mutex<Option<Vec<Credentials>>>,
141 transport: Mutex<Option<OpenedRoute>>,
142 pub state: Mutex<String>,
143 pub handle: Mutex<Option<Arc<Handle<ClientHandler>>>>,
144 shell: Mutex<Option<mpsc::UnboundedSender<ShellCommand>>>,
145 pub latency: Mutex<Option<Value>>,
146 pub finalized: AtomicBool,
147 lifecycle: Mutex<()>,
148 lost_tx: watch::Sender<bool>,
149 pub platform: Mutex<Option<Value>>,
150 pub privilege: Mutex<Option<Value>>,
151 pub sudo_cache: Mutex<Option<(Zeroizing<String>, Instant)>>,
152 pub windows_queue: tokio::sync::Mutex<()>,
153 pub metrics_state: Mutex<Value>,
154 pub stop_latency: Mutex<Option<tokio::task::JoinHandle<()>>>,
157impl SessionRecord {
158 pub fn has_jump(&self) -> bool {
159 self.route.len() > 1
160 }
162 pub fn lost(&self) -> watch::Receiver<bool> {
163 self.lost_tx.subscribe()
164 }
166 pub fn is_connected(&self) -> bool {
167 !self.finalized.load(Ordering::SeqCst) && *self.state.lock() == "connected"
168 }
170 pub fn handle(&self) -> CoreResult<Arc<Handle<ClientHandler>>> {
171 self.handle.lock().clone().ok_or_else(not_connected)
172 }
174 pub fn public_status(&self) -> Value {
175 json!({
176 "sessionId": self.session_id,
177 "profileId": self.profile_id,
178 "label": self.label,
179 "host": self.host,
180 "port": self.port,
181 "username": self.username,
182 "authMethod": self.auth_method,
183 "state": *self.state.lock(),
184 "slot": self.slot,
185 "adhoc": self.adhoc,
186 "jump": self.route.iter().take(self.route.len().saturating_sub(1)).map(|p| json!({
187 "label": p["label"], "host": p["host"], "port": p["port"], "username": p["username"]
188 })).collect::<Vec<_>>(),
189 "latency": self.latency.lock().clone().unwrap_or_else(|| json!({ "pingMs": null, "sshRttMs": null, "measuredAt": null }))
190 })
191 }
194pub fn not_connected() -> CoreError {
195 CoreError::new("SESSION_NOT_CONNECTED", "The terminal session is not connected.")
198#[derive(Clone)]
199struct PendingTrust {
200 profile_id: String,
201 host: String,
202 port: u16,
203 fingerprint: String,
204 known: Option<String>,
205 expires: Instant,
208pub type Emitter = Arc<dyn Fn(&str, Value) + Send + Sync>;
210#[derive(Clone)]
211enum Credentials {
212 Password(Zeroizing<String>),
213 Key(Arc<russh::keys::PrivateKey>),
216// The lease closes every opened transport on errors, cancellation and normal teardown.
217// Reverse order always disconnects the target before the jumps.
218#[derive(Default)]
219struct OpenedRoute {
220 handles: Vec<Arc<Handle<ClientHandler>>>,
221 address: String,
224impl Drop for OpenedRoute {
225 fn drop(&mut self) {
226 let handles = std::mem::take(&mut self.handles);
227 if !handles.is_empty() {
228 // No runtime exists during process shutdown; the OS then closes the sockets.
229 let Ok(runtime) = tokio::runtime::Handle::try_current() else { return };
230 runtime.spawn(async move {
231 for handle in handles.into_iter().rev() {
232 let _ = tokio::time::timeout(Duration::from_secs(2), handle.disconnect(russh::Disconnect::ByApplication, "", "en")).await;
233 }
234 });
235 }
236 }
239/// Internal probe snapshot. Never serialized or exposed over IPC. A Windows sshd
240/// restart may close the original session before its guarded login probe runs.
241pub struct ConnectionTarget {
242 route: Vec<Value>,
243 jump_credentials: Vec<Credentials>,
244 address: String,
247pub struct SessionManager {
248 pub profiles: Arc<ProfileService>,
249 pub logger: Arc<Logger>,
250 emit: Emitter,
251 sessions: Mutex<HashMap<String, Arc<SessionRecord>>>,
252 pending_trust: Mutex<HashMap<String, PendingTrust>>,
253 /// Internal subscribers (Transfer Manager) receive every public state change.
254 pub state_events: broadcast::Sender<Value>,
257impl SessionManager {
258 pub fn new(profiles: Arc<ProfileService>, logger: Arc<Logger>, emit: Emitter) -> Arc<Self> {
259 let (state_events, _) = broadcast::channel(256);
260 Arc::new(Self { profiles, logger, emit, sessions: Mutex::new(HashMap::new()), pending_trust: Mutex::new(HashMap::new()), state_events })
261 }
263 fn emit_state(&self, status: Value) {
264 let _ = self.state_events.send(status.clone());
265 (self.emit)("ssh:state", status);
266 }
268 /// Re-publishes the state of a live session (latency updates).
269 pub fn publish_state(&self, record: &Arc<SessionRecord>) {
270 let current = self.sessions.lock().get(&record.session_id).map(|r| Arc::ptr_eq(r, record)).unwrap_or(false);
271 if current && !record.finalized.load(Ordering::SeqCst) {
272 self.emit_state(record.public_status());
273 }
274 }
276 pub fn list(&self) -> Vec<Value> {
277 self.sessions.lock().values().map(|r| r.public_status()).collect()
278 }
280 pub fn get(&self, session_id: &str) -> Option<Arc<SessionRecord>> {
281 self.sessions.lock().get(session_id).cloned()
282 }
284 pub fn require_connected(&self, session_id: &str) -> CoreResult<Arc<SessionRecord>> {
285 let record = self.get(session_id).ok_or_else(not_connected)?;
286 if !record.is_connected() || record.shell.lock().is_none() {
287 return Err(not_connected());
288 }
289 Ok(record)
290 }
292 fn credentials(&self, profile: &Value, options: &Value) -> CoreResult<Credentials> {
293 self.credentials_fields(profile, options.get("password"), options.get("passphrase"))
294 }
296 fn credentials_fields(&self, profile: &Value, password: Option<&Value>, passphrase: Option<&Value>) -> CoreResult<Credentials> {
297 if profile["authMethod"] == "privateKey" {
298 let path = profile["privateKeyPath"].as_str().unwrap_or("");
299 let text = Zeroizing::new(read_private_key_file(path)?);
300 let passphrase = passphrase.and_then(Value::as_str).map(|s| Zeroizing::new(s.to_string()));
301 let key = decode_private_key(&text, passphrase.as_deref().map(|s| s.as_str()))?;
302 Ok(Credentials::Key(Arc::new(key)))
303 } else {
304 let id = profile["id"].as_str().unwrap_or("");
305 let password = password
306 .and_then(Value::as_str)
307 .filter(|p| !p.is_empty())
308 .map(str::to_string)
309 .or_else(|| if id.starts_with("adhoc:") { None } else { self.profiles.get_stored_password(id).ok().flatten() })
310 .unwrap_or_default();
311 if password.is_empty() {
312 return Err(CoreError::new("PASSWORD_REQUIRED", "Enter the SSH password."));
313 }
314 Ok(Credentials::Password(Zeroizing::new(password)))
315 }
316 }
318 async fn resolve_address(&self, host: &str, port: u16) -> CoreResult<IpAddr> {
319 if let Ok(ip) = host.parse::<IpAddr>() {
320 return Ok(ip);
321 }
322 let addresses: Vec<SocketAddr> = match tokio::net::lookup_host((host, port)).await {
323 Ok(list) => list.collect(),
324 Err(_) => Vec::new(),
325 };
326 let mut ips: Vec<IpAddr> = addresses.iter().map(|a| a.ip()).collect();
327 let mut source = "system";
328 if ips.is_empty() {
329 ips = crate::core::diagnostics::dns::resolve_both(host).await;
330 source = "dns";
331 }
332 let selected = ips.iter().find(|ip| ip.is_ipv4()).or_else(|| ips.first()).copied().ok_or_else(|| {
333 CoreError::new("SSH_DNS_RESOLUTION_FAILED", format!("DNS name \"{host}\" could not be resolved.")).with("host", host)
334 })?;
335 self.logger.info(
336 "SSH hostname resolved.",
337 json!({ "host": host, "address": selected.to_string(), "addressFamily": if selected.is_ipv4() { 4 } else { 6 }, "addressCount": ips.len(), "source": source }),
338 );
339 Ok(selected)
340 }
342 fn normalize_error(&self, error: &russh::Error, details: &[(&str, Value)]) -> CoreError {
343 let message = error.to_string();
344 let mut code = "SSH_CONNECTION_FAILED";
345 if let russh::Error::IO(io) = error {
346 let raw = io.raw_os_error();
347 if cfg!(target_os = "macos") && matches!(raw, Some(1) | Some(13)) {
348 code = "SSH_LOCAL_NETWORK_DENIED";
349 } else if cfg!(target_os = "macos") && matches!(raw, Some(65) | Some(51)) {
350 code = "SSH_MAC_NETWORK_UNREACHABLE";
351 } else if io.kind() == std::io::ErrorKind::ConnectionRefused {
352 code = "SSH_CONNECTION_REFUSED";
353 } else if io.kind() == std::io::ErrorKind::TimedOut {
354 code = "SSH_TIMEOUT";
355 }
356 } else if matches!(error, russh::Error::ConnectionTimeout | russh::Error::KeepaliveTimeout | russh::Error::InactivityTimeout | russh::Error::Elapsed(_)) {
357 code = "SSH_TIMEOUT";
358 } else if matches!(error, russh::Error::NotAuthenticated | russh::Error::NoAuthMethod) {
359 code = "SSH_AUTHENTICATION_FAILED";
360 }
361 let mut result = CoreError::new(code, message);
362 for (key, value) in details {
363 result = result.with(key, value.clone());
364 }
365 result
366 }
368 fn manual_profile(&self, input: &Value) -> CoreResult<Value> {
369 if !input.is_object() || !matches!(input["authMethod"].as_str(), Some("password" | "privateKey")) {
370 return Err(CoreError::new("SSH_TARGET_INVALID", "Invalid manual SSH target."));
371 }
372 let mut profile = normalize_profile(input, None)
373 .map_err(|e| CoreError::new("SSH_TARGET_INVALID", e.message))?;
374 profile["id"] = json!(format!("adhoc:{}", uuid::Uuid::new_v4()));
375 profile["label"] = json!(format!("{}@{}", profile["username"].as_str().unwrap_or(""), profile["host"].as_str().unwrap_or("")));
376 profile["jumpProfileId"] = json!("");
377 Ok(profile)
378 }
380 fn route_profiles(&self, options: &Value) -> CoreResult<Vec<Value>> {
381 let profile = match (options.get("profileId"), options.get("adhoc")) {
382 (Some(Value::String(id)), None) if !id.is_empty() => self.profiles.get(id)?,
383 (None, Some(input)) => self.manual_profile(input)?,
384 _ => return Err(CoreError::new("SSH_TARGET_INVALID", "Choose exactly one saved or manual SSH target.")),
385 };
386 let mut seen = std::collections::HashSet::from([profile["id"].as_str().unwrap_or("").to_string()]);
387 let saved_jump = |id: &str| self.profiles.get(id).map_err(|_| CoreError::new("JUMP_PROFILE_NOT_FOUND", "Jump profile was not found.")
388 .with("hop", "jump").with("hopHost", "").with("jumpProfileId", id));
389 let mut next = match options.get("jump") {
390 Some(Value::Null) => None,
391 None => profile["jumpProfileId"].as_str().filter(|id| !id.is_empty()).map(saved_jump).transpose()?,
392 Some(jump) if jump.is_object() && jump.get("profileId").is_some() => {
393 if jump.get("host").is_some() {
394 return Err(CoreError::new("SSH_TARGET_INVALID", "Choose one jump host."));
395 }
396 Some(saved_jump(jump["profileId"].as_str().unwrap_or(""))?)
397 }
398 Some(jump) => Some(self.manual_profile(jump)?),
399 };
400 let mut jumps = Vec::new();
401 while let Some(hop) = next {
402 if !seen.insert(hop["id"].as_str().unwrap_or("").to_string()) {
403 return Err(self.hop_error(CoreError::new("JUMP_LOOP", "The jump route contains a cycle."), &hop, true));
404 }
405 if jumps.len() == 3 {
406 return Err(self.hop_error(CoreError::new("JUMP_TOO_DEEP", "At most three jump hosts are allowed."), &hop, true));
407 }
408 next = hop["jumpProfileId"].as_str().filter(|id| !id.is_empty()).map(saved_jump).transpose()?;
409 jumps.push(hop);
410 }
411 jumps.reverse();
412 jumps.push(profile);
413 Ok(jumps)
414 }
416 fn hop_error(&self, error: CoreError, profile: &Value, jump: bool) -> CoreError {
417 let mut error = error.with("host", profile["host"].clone()).with("port", profile["port"].clone())
418 .with("hop", if jump { "jump" } else { "target" });
419 if jump {
420 error = error.with("hopHost", profile["host"].clone());
421 }
422 error
423 }
425 fn route_credentials(&self, route: &[Value], options: &Value) -> CoreResult<Vec<Credentials>> {
426 route[..route.len() - 1].iter().enumerate().map(|(i, hop)| {
427 let (password, passphrase) = if i + 2 == route.len() {
428 (options.get("jumpPassword"), options.get("jumpPassphrase"))
429 } else { (None, None) };
430 self.credentials_fields(hop, password, passphrase).map_err(|e| self.hop_error(e, hop, true))
431 }).collect()
432 }
434 /// Shared handshake, host-key verification and authentication for terminals and
435 /// fresh logins. Only the first transport is resolved locally.
436 async fn open_route(&self, route: &[Value], jumps: &[Credentials], target: &Credentials, address: Option<&str>) -> CoreResult<OpenedRoute> {
437 let mut opened = OpenedRoute::default();
438 for (i, profile) in route.iter().enumerate() {
439 let jump = i + 1 < route.len();
440 let host = profile["host"].as_str().unwrap_or("");
441 let port = profile["port"].as_u64().unwrap_or(22) as u16;
442 let known = self.profiles.get_host_key(host, port).map_err(|e| self.hop_error(e, profile, jump))?;
443 let observed = Arc::new(Mutex::new(None));
444 let handler = ClientHandler { expected: known.clone(), observed: observed.clone() };
445 let creds = if jump { &jumps[i] } else { target };
446 let attempt = tokio::time::timeout(READY_TIMEOUT, async {
447 let mut handle = if let Some(previous) = opened.handles.last() {
448 let channel = previous.channel_open_direct_tcpip(host, u32::from(port), "127.0.0.1", 0).await
449 .map_err(|_| ConnectFailure::Forward)?;
450 client::connect_stream(client_config(true), channel.into_stream(), handler).await.map_err(ConnectFailure::Ssh)?
451 } else {
452 let ip = match address {
453 Some(address) => address.parse::<IpAddr>().map_err(|_| ConnectFailure::Core(CoreError::new("SSH_CONNECTION_FAILED", "Invalid first-hop address.")))?,
454 None => self.resolve_address(host, port).await.map_err(ConnectFailure::Core)?,
455 };
456 opened.address = ip.to_string();
457 client::connect(client_config(true), SocketAddr::new(ip, port), handler).await.map_err(ConnectFailure::Ssh)?
458 };
459 let result = match creds {
460 Credentials::Password(password) => handle.authenticate_password(profile["username"].as_str().unwrap_or(""), password.as_str()).await,
461 Credentials::Key(key) => {
462 let hash = handle.best_supported_rsa_hash().await.map_err(ConnectFailure::Ssh)?.flatten();
463 handle.authenticate_publickey(profile["username"].as_str().unwrap_or(""), PrivateKeyWithHashAlg::new(key.clone(), hash)).await
464 }
465 }.map_err(ConnectFailure::Ssh)?;
466 if !result.success() {
467 let _ = handle.disconnect(russh::Disconnect::ByApplication, "", "en").await;
468 return Err(ConnectFailure::Auth);
469 }
470 Ok::<_, ConnectFailure>(handle)
471 }).await;
472 match attempt {
473 Ok(Ok(handle)) => opened.handles.push(Arc::new(handle)),
474 failure => {
475 let fingerprint = observed.lock().clone();
476 let error = match failure {
477 Ok(Err(ConnectFailure::Ssh(russh::Error::UnknownKey))) if fingerprint.is_some() => {
478 let mut error = CoreError::new(if known.is_some() { "HOST_KEY_MISMATCH" } else { "HOST_KEY_UNKNOWN" }, "SSH host key verification requires confirmation.")
479 .with("fingerprint", fingerprint.unwrap());
480 if let Some(known) = known { error = error.with("knownFingerprint", known); }
481 error
482 }
483 Ok(Err(ConnectFailure::Forward)) => {
484 // The refusal belongs to the forwarding jump, not the next host.
485 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));
486 }
487 Ok(Err(ConnectFailure::Ssh(e))) => self.normalize_error(&e, &[]),
488 Ok(Err(ConnectFailure::Auth)) => CoreError::new("SSH_AUTHENTICATION_FAILED", "All configured authentication methods failed."),
489 Ok(Err(ConnectFailure::Core(e))) => e,
490 Err(_) => CoreError::new("SSH_TIMEOUT", "Timed out while waiting for handshake."),
491 Ok(Ok(_)) => unreachable!(),
492 };
493 return Err(self.hop_error(error, profile, jump));
494 }
495 }
496 }
497 Ok(opened)
498 }
500 pub async fn connect(self: &Arc<Self>, options: &Value) -> CoreResult<Value> {
501 let session_id = options.get("sessionId").and_then(Value::as_str).unwrap_or("").trim().to_string();
502 if !SESSION_ID.is_match(&session_id) {
503 return Err(CoreError::new("INVALID_SESSION_ID", "Invalid terminal session ID."));
504 }
505 let slot = match options.get("slot") {
506 None | Some(Value::Null) => None,
507 Some(v) if v.as_u64() == Some(1) || v.as_u64() == Some(2) => Some(v.as_u64().unwrap() as u8),
508 _ => return Err(CoreError::new("SSH_TARGET_INVALID", "Invalid terminal slot.")),
509 };
510 let route = self.route_profiles(options)?;
511 let profile = route.last().unwrap().clone();
512 let jumps = self.route_credentials(&route, options)?;
513 let credentials = self.credentials(&profile, options).map_err(|e| self.hop_error(e, &profile, false))?;
514 let (lost_tx, _) = watch::channel(false);
515 let record = Arc::new(SessionRecord {
516 session_id: session_id.clone(),
517 profile_id: profile["id"].as_str().unwrap_or("").to_string(),
518 label: profile["label"].as_str().unwrap_or("").to_string(),
519 host: profile["host"].as_str().unwrap_or("").to_string(),
520 address: Mutex::new(profile["host"].as_str().unwrap_or("").to_string()),
521 port: profile["port"].as_u64().unwrap_or(22) as u16,
522 username: profile["username"].as_str().unwrap_or("").to_string(),
523 auth_method: profile["authMethod"].as_str().unwrap_or("password").to_string(),
524 slot,
525 adhoc: profile["id"].as_str().unwrap_or("").starts_with("adhoc:"),
526 profile,
527 route,
528 jump_credentials: Mutex::new(None),
529 transport: Mutex::new(None),
530 state: Mutex::new("connecting".into()),
531 handle: Mutex::new(None),
532 shell: Mutex::new(None),
533 latency: Mutex::new(None),
534 finalized: AtomicBool::new(false),
535 lifecycle: Mutex::new(()),
536 lost_tx,
537 platform: Mutex::new(None),
538 privilege: Mutex::new(None),
539 sudo_cache: Mutex::new(None),
540 windows_queue: tokio::sync::Mutex::new(()),
541 metrics_state: Mutex::new(json!({})),
542 stop_latency: Mutex::new(None),
543 });
544 let previous = {
545 let mut sessions = self.sessions.lock();
546 self.pending_trust.lock().remove(&session_id);
547 sessions.insert(session_id.clone(), record.clone())
548 };
549 if let Some(previous) = previous { self.finalize(&previous, "reconnect"); }
550 self.emit_state(record.public_status());
551 let cols = normalize_terminal_size(options.get("cols"), 100, 20, 500);
552 let rows = normalize_terminal_size(options.get("rows"), 30, 5, 300);
553 let mut lost = record.lost();
554 let result = {
555 let attempt = async {
556 let transport = self.open_route(&record.route, &jumps, &credentials, None).await?;
557 let handle = transport.handles.last().unwrap().clone();
558 let channel = tokio::time::timeout(READY_TIMEOUT, async {
559 let channel = handle.channel_open_session().await?;
560 channel.request_pty(false, "xterm-256color", cols, rows, 0, 0, &[]).await?;
561 channel.request_shell(true).await?;
562 Ok::<_, russh::Error>(channel)
563 }).await.map_err(|_| CoreError::new("SSH_TIMEOUT", "Timed out opening the terminal."))?
564 .map_err(|e| self.normalize_error(&e, &[]))?;
565 Ok::<_, CoreError>((transport, handle, channel))
566 };
567 let cancelled = *lost.borrow();
568 if cancelled { Err(not_connected()) } else {
569 tokio::select! {
570 result = attempt => result,
571 _ = lost.changed() => Err(not_connected()),
572 }
573 }
574 };
575 match result {
576 Ok((transport, handle, channel)) => {
577 let status = {
578 let _lifecycle = record.lifecycle.lock();
579 let sessions = self.sessions.lock();
580 let current = sessions.get(&session_id).is_some_and(|r| Arc::ptr_eq(r, &record));
581 if !current || record.finalized.load(Ordering::SeqCst) { return Err(not_connected()); }
582 self.pending_trust.lock().remove(&session_id);
583 drop(sessions);
584 if !record.has_jump() { *record.address.lock() = transport.address.clone(); }
585 *record.handle.lock() = Some(handle);
586 *record.transport.lock() = Some(transport);
587 *record.jump_credentials.lock() = Some(jumps);
588 *record.state.lock() = "connected".into();
589 self.bind_shell(&record, channel);
590 if !record.adhoc { let _ = self.profiles.mark_used(&record.profile_id); }
591 record.public_status()
592 };
593 self.logger.info("SSH session connected.", json!({ "sessionId": record.session_id, "profileId": record.profile_id, "host": record.host, "port": record.port, "username": record.username, "authMethod": record.auth_method }));
594 self.publish_state(&record);
595 crate::core::ssh::latency::start(self.clone(), record.clone());
596 let manager = self.clone();
597 tokio::spawn(async move {
598 let mut lost = record.lost();
599 loop {
600 tokio::select! {
601 _ = lost.changed() => break,
602 _ = tokio::time::sleep(Duration::from_millis(250)) => {
603 let closed = record.transport.lock().as_ref().map(|t| t.handles.iter().any(|h| h.is_closed())).unwrap_or(true);
604 if closed { manager.finalize(&record, "route-close"); break; }
605 }
606 }
607 }
608 });
609 Ok(status)
610 }
611 Err(error) => {
612 let error = error.with("sessionId", session_id.clone()).with("profileId", if record.adhoc { "" } else { &record.profile_id });
613 let lifecycle = record.lifecycle.lock();
614 let sessions = self.sessions.lock();
615 let current = sessions.get(&record.session_id).is_some_and(|r| Arc::ptr_eq(r, &record));
616 if current && !record.finalized.load(Ordering::SeqCst) && matches!(error.code.as_str(), "HOST_KEY_UNKNOWN" | "HOST_KEY_MISMATCH") {
617 self.pending_trust.lock().insert(session_id, PendingTrust {
618 profile_id: if record.adhoc { String::new() } else { record.profile_id.clone() },
619 host: error.details["host"].as_str().unwrap_or("").to_string(),
620 port: error.details["port"].as_u64().unwrap_or(22) as u16,
621 fingerprint: error.details["fingerprint"].as_str().unwrap_or("").to_string(),
622 known: error.details.get("knownFingerprint").and_then(Value::as_str).map(str::to_string),
623 expires: Instant::now() + PENDING_TRUST_TTL,
624 });
625 }
626 drop(sessions);
627 drop(lifecycle);
628 self.logger.warn("SSH connection failed.", json!({ "sessionId": record.session_id, "code": error.code, "message": error.message, "hop": error.details.get("hop"), "host": error.details.get("host"), "port": error.details.get("port") }));
629 self.finalize(&record, &error.code);
630 Err(error)
631 }
632 }
633 }
635 fn bind_shell(self: &Arc<Self>, record: &Arc<SessionRecord>, channel: russh::Channel<Msg>) {
636 let (tx, mut rx) = mpsc::unbounded_channel::<ShellCommand>();
637 *record.shell.lock() = Some(tx);
638 let (mut read, write): (_, ChannelWriteHalf<Msg>) = channel.split();
639 let manager = self.clone();
640 let record = record.clone();
641 tokio::spawn(async move {
642 let mut decoder = Utf8Decoder::default();
643 let mut lost = record.lost();
644 let reason = loop {
645 if !record.is_connected() { break "disconnected"; }
646 tokio::select! {
647 message = read.wait() => match message {
648 Some(ChannelMsg::Data { data }) => {
649 let text = decoder.write(&data);
650 if !text.is_empty() && record.is_connected() { (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": text })); }
651 }
652 Some(ChannelMsg::ExtendedData { data, .. }) => {
653 let text = decoder.write(&data);
654 if !text.is_empty() && record.is_connected() { (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": text })); }
655 }
656 Some(ChannelMsg::Close) | None => break "shell-close",
657 Some(ChannelMsg::Failure) => {
658 (manager.emit)("ssh:error", json!({ "sessionId": record.session_id, "message": "The SSH server refused the interactive shell." }));
659 break "shell-error";
660 }
661 Some(_) => {}
662 },
663 command = rx.recv() => match command {
664 Some(ShellCommand::Input(bytes)) => { if write.data_bytes(bytes).await.is_err() { break "shell-error"; } }
665 Some(ShellCommand::Resize(cols, rows)) => { let _ = write.window_change(cols, rows, 0, 0).await; }
666 None => break "shell-close",
667 },
668 _ = lost.changed() => break "disconnected",
669 }
670 };
671 let trailing = decoder.end();
672 if !trailing.is_empty() && record.is_connected() {
673 (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": trailing }));
674 }
675 let _ = write.close().await;
676 manager.finalize(&record, reason);
677 });
678 }
680 pub fn trust_pending(&self, options: &Value) -> CoreResult<Value> {
681 let session_id = options.get("sessionId").and_then(Value::as_str).unwrap_or("").to_string();
682 let fingerprint = options.get("fingerprint").and_then(Value::as_str).unwrap_or("").to_lowercase();
683 let pending = self.pending_trust.lock().get(&session_id).cloned();
684 let Some(pending) = pending.filter(|p| p.expires >= Instant::now()) else {
685 self.pending_trust.lock().remove(&session_id);
686 return Err(CoreError::new("HOST_KEY_CONFIRMATION_EXPIRED", "The host-key confirmation expired."));
687 };
688 if options.get("profileId").and_then(Value::as_str) != Some(pending.profile_id.as_str()) || pending.fingerprint != fingerprint {
689 return Err(CoreError::new("HOST_KEY_CONFIRMATION_INVALID", "The host-key confirmation does not match the pending connection."));
690 }
691 let replace = options.get("replaceKnown") == Some(&Value::Bool(true));
692 if let Some(known) = &pending.known {
693 let confirmed = options.get("knownFingerprint").and_then(Value::as_str).unwrap_or("").to_lowercase();
694 if !replace || &confirmed != known {
695 return Err(CoreError::new(
696 "HOST_KEY_REPLACEMENT_CONFIRMATION_REQUIRED",
697 "Replacing a changed SSH host key requires explicit confirmation of both fingerprints.",
698 ));
699 }
700 if self.profiles.get_host_key(&pending.host, pending.port)?.as_deref() != Some(known.as_str()) {
701 self.pending_trust.lock().remove(&session_id);
702 return Err(CoreError::new("HOST_KEY_REPLACEMENT_STALE", "The stored SSH host key changed while confirmation was pending."));
703 }
704 } else if replace {
705 return Err(CoreError::new("HOST_KEY_CONFIRMATION_INVALID", "A new host key cannot be confirmed as a replacement."));
706 }
707 let result = self.profiles.trust_host_key(&pending.host, pending.port, &pending.fingerprint)?;
708 self.pending_trust.lock().remove(&session_id);
709 let details = json!({ "host": pending.host, "port": pending.port, "fingerprint": pending.fingerprint, "replacedFingerprint": pending.known });
710 if pending.known.is_some() {
711 self.logger.warn("Changed SSH host key replaced by user confirmation.", details);
712 } else {
713 self.logger.info("SSH host key trusted by user.", details);
714 }
715 Ok(result)
716 }
718 /// Terminal input. Called synchronously and in order; data is queued to the shell task.
719 pub fn write(&self, session_id: &str, data: &str) -> CoreResult<()> {
720 let record = self.require_connected(session_id)?;
721 if data.len() > MAX_INPUT_LENGTH {
722 return Err(CoreError::new("TERMINAL_INPUT_TOO_LARGE", "Terminal input is too large."));
723 }
724 let sender = record.shell.lock().clone().ok_or_else(not_connected)?;
725 sender.send(ShellCommand::Input(data.as_bytes().to_vec())).map_err(|_| not_connected())
726 }
728 pub fn resize(&self, session_id: &str, cols: Option<&Value>, rows: Option<&Value>) -> CoreResult<Value> {
729 let record = self.require_connected(session_id)?;
730 let cols = normalize_terminal_size(cols, 100, 20, 500);
731 let rows = normalize_terminal_size(rows, 30, 5, 300);
732 if let Some(sender) = record.shell.lock().clone() {
733 let _ = sender.send(ShellCommand::Resize(cols, rows));
734 }
735 Ok(json!({ "cols": cols, "rows": rows }))
736 }
738 /// The fingerprint this session's server is trusted with now.
739 pub fn trusted_host_key(&self, session_id: &str) -> Option<String> {
740 let record = self.get(session_id)?;
741 self.profiles.get_host_key(&record.host, record.port).ok().flatten()
742 }
744 /// Availability of a jump-only server, through a session already open to its jump host:
745 /// one forwarding channel, no login and no new session. Never an unproven "online".
746 pub async fn probe_through_jump(&self, jump_profile_id: &str, host: &str, port: u16, timeout: Duration) -> Outcome {
747 let handle = self.sessions.lock().values()
748 .find(|record| record.profile_id == jump_profile_id && record.state.lock().as_str() == "connected")
749 .and_then(|record| record.handle.lock().clone());
750 let Some(handle) = handle else { return Outcome::Unverified { reason: "noJumpSession" } };
751 let start = std::time::Instant::now();
752 match tokio::time::timeout(timeout, handle.channel_open_direct_tcpip(host, u32::from(port), "127.0.0.1", 0)).await {
753 Ok(Ok(channel)) => {
754 let ms = (start.elapsed().as_secs_f64() * 10_000.0).round() / 10.0;
755 let _ = channel.close().await;
756 Outcome::Online { ms }
757 }
758 Ok(Err(russh::Error::ChannelOpenFailure(russh::ChannelOpenFailure::ConnectFailed))) => Outcome::Failed { reason: "refused" },
759 Ok(Err(_)) => Outcome::Unverified { reason: "jumpDenied" },
760 Err(_) => Outcome::Failed { reason: "timeout" },
761 }
762 }
764 pub async fn disconnect(&self, session_id: &str, reason: &str) -> Value {
765 self.pending_trust.lock().remove(session_id);
766 let record = self.sessions.lock().get(session_id).cloned();
767 match record {
768 Some(record) => {
769 self.finalize(&record, reason);
770 json!({ "disconnected": true })
771 }
772 None => json!({ "disconnected": false }),
773 }
774 }
776 pub async fn disconnect_all(&self, reason: &str) -> Value {
777 let ids: Vec<String> = self.sessions.lock().keys().cloned().collect();
778 for id in &ids {
779 self.disconnect(id, reason).await;
780 }
781 self.pending_trust.lock().clear();
782 json!({ "disconnected": ids.len() })
783 }
785 pub fn finalize(&self, record: &Arc<SessionRecord>, reason: &str) {
786 let lifecycle = record.lifecycle.lock();
787 if record.finalized.swap(true, Ordering::SeqCst) {
788 return;
789 }
790 *record.state.lock() = "disconnected".into();
791 if let Some(task) = record.stop_latency.lock().take() {
792 task.abort();
793 }
794 *record.latency.lock() = None;
795 record.lost_tx.send_replace(true);
796 *record.sudo_cache.lock() = None;
797 {
798 let mut sessions = self.sessions.lock();
799 if sessions.get(&record.session_id).map(|r| Arc::ptr_eq(r, record)).unwrap_or(false) {
800 sessions.remove(&record.session_id);
801 }
802 }
803 record.shell.lock().take();
804 record.handle.lock().take();
805 record.transport.lock().take();
806 record.jump_credentials.lock().take();
807 let mut status = record.public_status();
808 status["reason"] = Value::String(reason.to_string());
809 drop(lifecycle);
810 self.emit_state(status);
811 self.logger.info("SSH session disconnected.", json!({ "sessionId": record.session_id, "profileId": record.profile_id, "reason": reason }));
812 }
814 // -----------------------------------------------------------------------------------------
815 // Exec channels
816 // -----------------------------------------------------------------------------------------
818 pub async fn exec_channel(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration, limit: usize) -> CoreResult<ExecChannel> {
819 let handle = record.handle()?;
820 let lost = record.lost();
821 let opened = tokio::time::timeout(timeout, async {
822 let channel = handle.channel_open_session().await?;
823 channel.exec(true, command.as_bytes().to_vec()).await?;
824 Ok::<_, russh::Error>(channel)
825 })
826 .await
827 .map_err(|_| CoreError::new("REMOTE_COMMAND_TIMEOUT", "The remote command timed out."))?
828 .map_err(|e| if record.is_connected() { CoreError::new("REMOTE_COMMAND_FAILED", e.to_string()) } else { not_connected() })?;
829 let (read, write) = opened.split();
830 Ok(ExecChannel::new(read, write, timeout, limit, lost))
831 }
833 /// Runs a fixed command and requires exit code 0 (`_execFixed`).
834 pub async fn exec_fixed(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration, limit: usize) -> CoreResult<ExecResult> {
835 let mut channel = self.exec_channel(record, command, timeout, limit).await?;
836 channel.eof().await;
837 let result = channel.finish().await?;
838 match result.code {
839 Some(0) => Ok(result),
840 code => {
841 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() };
842 Err(CoreError::new("REMOTE_COMMAND_FAILED", message).with("remoteExitCode", code.map(Value::from).unwrap_or(Value::Null)))
843 }
844 }
845 }
847 pub async fn exec_fixed_default(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration) -> CoreResult<ExecResult> {
848 self.exec_fixed(record, command, timeout, MAX_COMMAND_OUTPUT).await
849 }
851 /// One transport round trip (keepalive@openssh.com with want-reply), fallback `exit 0`.
852 pub async fn ssh_round_trip(&self, record: &Arc<SessionRecord>) -> CoreResult<(Option<f64>, &'static str)> {
853 let handle = record.handle()?;
854 let started = Instant::now();
855 match tokio::time::timeout(Duration::from_secs(5), handle.send_ping()).await {
856 Ok(Ok(())) => return Ok((Some(started.elapsed().as_secs_f64() * 1000.0), "transport")),
857 Ok(Err(_)) => {}
858 Err(_) => return Err(CoreError::new("SSH_RTT_TIMEOUT", "The SSH server did not answer the round-trip request in time.")),
859 }
860 let started = Instant::now();
861 self.exec_fixed_default(record, "exit 0", Duration::from_secs(5)).await?;
862 Ok((Some(started.elapsed().as_secs_f64() * 1000.0), "command"))
863 }
865 /// Serializes PowerShell helpers of one Windows session (Win32-OpenSSH stalls on concurrent stdin helpers).
866 pub async fn windows_serial<T, F: std::future::Future<Output = CoreResult<T>>>(&self, record: &Arc<SessionRecord>, task: F) -> CoreResult<T> {
867 let _turn = record.windows_queue.lock().await;
868 task.await
869 }
871 // -----------------------------------------------------------------------------------------
872 // Identity facts, fresh logins (Users lockout protection)
873 // -----------------------------------------------------------------------------------------
875 pub fn session_auth_info(&self, session_id: &str) -> CoreResult<Value> {
876 let record = self.require_connected(session_id)?;
877 let fingerprint = if record.auth_method == "privateKey" {
878 crate::core::ssh::keys::public_key_fingerprint_for_private_key(record.profile["privateKeyPath"].as_str().unwrap_or("")).map(Value::String).unwrap_or(Value::Null)
879 } else { Value::Null };
880 // The target address is not the SSH client's source address. Users policy
881 // obtains the actual source from the remote SSH_CONNECTION environment.
882 Ok(json!({ "username": record.username, "authMethod": record.auth_method, "address": record.address.lock().clone(), "publicKeyFingerprint": fingerprint }))
883 }
885 pub fn probe_login_requirements(&self, session_id: &str, credentials: &Value) -> CoreResult<Value> {
886 let record = self.require_connected(session_id)?;
887 Ok(match self.credentials(&record.profile, credentials) {
888 Ok(_) => json!({ "ready": true, "missing": null }),
889 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" } }),
890 })
891 }
893 pub fn connection_target(&self, session_id: &str) -> CoreResult<ConnectionTarget> {
894 let record = self.require_connected(session_id)?;
895 let _lifecycle = record.lifecycle.lock();
896 if !record.is_connected() { return Err(not_connected()); }
897 let jump_credentials = record.jump_credentials.lock().clone().ok_or_else(not_connected)?;
898 let address = record.address.lock().clone();
899 Ok(ConnectionTarget {
900 route: record.route.clone(),
901 jump_credentials,
902 address,
903 })
904 }
906 /// One fixed command over a new connection following the original verified route.
907 pub async fn fresh_exec(&self, target: &ConnectionTarget, credentials: &Value, command: &str, input: &str, timeout: Duration, accept_any_exit: bool) -> CoreResult<Value> {
908 let probe_failed = |error: CoreError| {
909 let mut wrapped = CoreError::new("SSH_PROBE_FAILED", format!("A new SSH login was not confirmed: {}", error.message));
910 wrapped.details = error.details;
911 wrapped.with("causeCode", error.code)
912 };
913 let run = async {
914 let profile = target.route.last().ok_or_else(not_connected)?;
915 let creds = self.credentials(profile, credentials)?;
916 let address = (target.route.len() == 1).then_some(target.address.as_str());
917 let transport = self.open_route(&target.route, &target.jump_credentials, &creds, address).await?;
918 let handle = transport.handles.last().unwrap();
919 let channel = handle.channel_open_session().await.map_err(|e| self.normalize_error(&e, &[]))?;
920 channel.exec(true, command.as_bytes().to_vec()).await.map_err(|e| self.normalize_error(&e, &[]))?;
921 let (read, write) = channel.split();
922 let (_tx, lost) = watch::channel(false);
923 let mut exec = ExecChannel::new(read, write, timeout, MAX_COMMAND_OUTPUT, lost);
924 if !input.is_empty() { exec.write(input.as_bytes()).await; }
925 exec.eof().await;
926 let result = exec.finish().await?;
927 if !accept_any_exit && !matches!(result.code, Some(0) | None) {
928 return Err(CoreError::new("SSH_PROBE_EXIT", format!("Probe command exited with {}.", result.code.unwrap_or(-1))));
929 }
930 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) }))
931 };
932 tokio::time::timeout(timeout, run).await.map_err(|_| CoreError::new("SSH_TIMEOUT", "Login probe timed out.")).and_then(|r| r).map_err(probe_failed)
933 }
935 pub async fn probe_login(&self, session_id: &str, credentials: &Value) -> CoreResult<Value> {
936 let target = self.connection_target(session_id)?;
937 let result = self.fresh_exec(&target, credentials, "true", "", Duration::from_secs(60), true).await?;
938 Ok(json!({ "ok": true, "address": result["address"], "checkedAt": result["checkedAt"] }))
939 }
942enum ConnectFailure {
943 Ssh(russh::Error),
944 Auth,
945 Forward,
946 Core(CoreError),
949#[cfg(test)]
950mod tests {
951 use super::*;
953 #[test]
954 fn terminal_size_bounds() {
955 assert_eq!(normalize_terminal_size(Some(&json!(10)), 100, 20, 500), 20);
956 assert_eq!(normalize_terminal_size(Some(&json!(9999)), 100, 20, 500), 500);
957 assert_eq!(normalize_terminal_size(Some(&json!(80.5)), 100, 20, 500), 100);
958 assert_eq!(normalize_terminal_size(None, 30, 5, 300), 30);
959 }
961 #[test]
962 fn quoting() {
963 assert_eq!(shell_quote("a'b"), "'a'\"'\"'b'");
964 }
967#[cfg(test)]
968#[path = "../../../tests/support/ssh_route_unit.rs"]
969mod route_tests;
SHA-256b8ed14a1c6450c1b8d83525f05aa36213cd5e6cb1141403aeec238bd051f94d0

CH-J Proprietary Software License 1.14 · کد منبع تحت CH-J Proprietary Software License 1.14 ارائه می‌شود. در دسترس بودن آن شرایط مجوز را تغییر نمی‌دهد و حق اضافه‌ای اعطا نمی‌کند.

SHA-256 بایگانی77380d4805485ffa4027fbe7858b5836f197a4aa9d52358a43e2d12d018744df
فایل‌ها