""" SSH Manager - manages SSH connections to VPN servers. Replicates the ServerController logic from the AmneziaVPN client. """ import paramiko import io import time import threading import logging logger = logging.getLogger(__name__) class SSHManager: """Manages SSH connections and command execution on remote servers.""" def __init__(self, host, port, username, password=None, private_key=None, connect_cooldown_base=30.0): self.host = host self.port = int(port) self.username = username self.password = password self.private_key = private_key self.client = None self._is_root = (username == 'root') # Serializes connect/disconnect so concurrent threads (UI request # handler + background monitor) cannot race a half-built transport. self._conn_lock = threading.Lock() # Serializes command/SFTP execution on the shared transport. Pooled # managers are used concurrently by request handlers and background # threads (traffic sync, conn monitor); without this a failed # exec on one side triggers a reconnect that closes the client the # other side is mid-command on ('NoneType open_session', EOFError # in open_sftp, bursts of 500s followed by the connect cooldown). # RLock because run_command retries by calling itself. self._exec_lock = threading.RLock() # Circuit breaker: after a failed connect, do not hammer the dead # server on every request (each attempt costs up to `timeout` seconds # and can exhaust the web worker pool when several servers are down). # Consecutive failures grow the cooldown base -> 2x -> 4x -> ... up # to 300s; the first successful connect resets it back to base. # Base is configurable per server (data.json: ssh_cooldown_base). self._connect_cooldown_base = float(connect_cooldown_base or 30.0) self._last_connect_fail = 0.0 self._connect_cooldown = self._connect_cooldown_base self._connect_fail_count = 0 self._connect_cooldown_max = 300.0 # Pooled managers (shared via app.get_ssh) must ignore the legacy # per-request disconnect() calls scattered across endpoints — # otherwise every API request kills the shared transport. self.pooled = False def connect(self): """Establish SSH connection to the server.""" with self._conn_lock: self._disconnect_locked() # One retry on TCP connect timeout: links with random SYN loss # (e.g. transcontinental/DPI-filtered routes) drop ~half of the # first attempts while the retry succeeds in milliseconds. last_exc = None for attempt in (1, 2): try: self._connect_once() last_exc = None break except (TimeoutError, OSError) as e: last_exc = e logger.warning( f"SSH connect to {self.host} attempt {attempt} " f"failed: {e}") self._disconnect_locked() if last_exc is not None: self._record_connect_failure() raise last_exc self._reset_connect_failures() return True def _record_connect_failure(self): """Feed the breaker from any failed connect path (caller holds _conn_lock): bump the streak and grow the cooldown exponentially.""" self._connect_fail_count += 1 self._last_connect_fail = time.time() self._connect_cooldown = min( self._connect_cooldown_base * (2 ** (self._connect_fail_count - 1)), self._connect_cooldown_max) def _reset_connect_failures(self): """First successful connect closes the breaker again.""" self._connect_fail_count = 0 self._connect_cooldown = self._connect_cooldown_base self._last_connect_fail = 0.0 def _connect_once(self): """Single TCP+SSH handshake attempt (caller holds _conn_lock).""" self.client = paramiko.SSHClient() self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) kwargs = { 'hostname': self.host, 'port': self.port, 'username': self.username, 'timeout': 15, 'allow_agent': False, 'look_for_keys': False, } if self.private_key: key_file = io.StringIO(self.private_key) try: pkey = paramiko.RSAKey.from_private_key(key_file) except paramiko.ssh_exception.SSHException: key_file.seek(0) try: pkey = paramiko.Ed25519Key.from_private_key(key_file) except paramiko.ssh_exception.SSHException: key_file.seek(0) pkey = paramiko.ECDSAKey.from_private_key(key_file) kwargs['pkey'] = pkey elif self.password: kwargs['password'] = self.password self.client.connect(**kwargs) # Keep NAT/stateful firewalls from silently dropping the idle # long-lived transport between command bursts. try: self.client.get_transport().set_keepalive(15) except Exception: pass def _disconnect_locked(self): """Close SSH connection (caller must hold _conn_lock).""" if self.client: try: self.client.close() except Exception: pass self.client = None def disconnect(self): """Close SSH connection. No-op for pooled managers: legacy endpoints call disconnect() in finally-blocks after every request, which would destroy the shared long-lived transport. Use force_disconnect() to really close it. """ if self.pooled: return with self._conn_lock: self._disconnect_locked() def force_disconnect(self): """Unconditionally close SSH connection (pool eviction etc.).""" with self._conn_lock: self._disconnect_locked() def ensure_connected(self): """Connect only if there is no live transport. Lets the panel keep one long-lived connection per server and run commands as cheap channels on it instead of paying a full TCP+SSH handshake for every API request (a major source of UI timeouts on high-latency servers). """ try: transport = self.client.get_transport() if self.client else None if transport and transport.is_active(): return True except Exception: pass # Cooldown after a recent failed attempt: fail fast instead of # blocking the worker on another 15s connect to a dead server. if time.time() - self._last_connect_fail < self._connect_cooldown: raise ConnectionError( f"SSH to {self.host} recently failed, backing off " f"{int(self._connect_cooldown)}s") # connect() itself feeds the breaker on failure and resets it on # success, so reconnect-and-retry paths are covered too. self.connect() return True def run_command(self, command, timeout=60, _retried=False, stdin_input=None): """Execute command on remote server. stdin_input, when given, is written to the channel's stdin right after exec and the write side is closed (same semantics as a shell pipe). Used to feed the sudo password without putting it on the command line. """ with self._exec_lock: return self._run_command_locked(command, timeout, _retried, stdin_input) @staticmethod def _reason(exc): """Text for an exception that often carries none: paramiko raises bare EOFError/SSHException when the transport dies mid-command, and an empty string travels all the way to the UI as "... failed: ".""" return str(exc).strip() or type(exc).__name__ def _run_command_locked(self, command, timeout, _retried, stdin_input=None): self.ensure_connected() logger.info(f"Running command: {command[:100]}...") try: stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout) except Exception as e: # Transport can be dead while is_active() still claims otherwise # (silent NAT drop). Reconnect once and retry before giving up. if not _retried: logger.warning(f"exec failed ({e}); reconnecting and retrying once") try: self.connect() except Exception as ce: logger.error(f"reconnect failed: {ce}") return "", self._reason(ce), -1 return self.run_command(command, timeout=timeout, _retried=True, stdin_input=stdin_input) logger.error(f"exec failed after retry: {e}") return "", self._reason(e), -1 if stdin_input is not None: try: stdin.write(stdin_input) stdin.flush() # Signal EOF so consumers of stdin (e.g. docker exec -i) # behave exactly as with the old `echo ... |` pipe. stdin.channel.shutdown_write() except Exception as e: logger.warning(f"failed to write command stdin: {e}") # Crucial: set timeout on the channel to prevent hanging indefinitely stdout.channel.settimeout(timeout) stderr.channel.settimeout(timeout) try: exit_code = stdout.channel.recv_exit_status() out = stdout.read().decode('utf-8', errors='replace').strip() err = stderr.read().decode('utf-8', errors='replace').strip() except Exception as e: logger.error(f"Command timed out or failed to read: {e}") out, err, exit_code = "", f"SSH connection lost while running the command ({self._reason(e)})", -1 if exit_code != 0: logger.warning(f"Command exited with code {exit_code}: {err}") return out, err, exit_code def run_sudo_command(self, command, timeout=60): """ Execute command with sudo, automatically handling password. Strips 'sudo ' from the beginning of command if present, and re-adds it with password piping. """ # Remove existing sudo prefix if present clean_cmd = command if clean_cmd.strip().startswith('sudo '): clean_cmd = clean_cmd.strip()[5:] if self._is_root: return self.run_command(clean_cmd, timeout=timeout) if self.password: # Feed the password via channel stdin: keeping it out of the # command line hides it from both the panel log (logged above) # and the remote process list. return self.run_command( f"sudo -S -p '' {clean_cmd}", timeout=timeout, stdin_input=self.password + '\n') return self.run_command(f"sudo {clean_cmd}", timeout=timeout) def run_sudo_script(self, script, timeout=120): """ Execute a multi-line script with sudo/root privileges. Writes script to /tmp via SFTP, then runs with sudo bash. """ if self._is_root: return self.run_script(script, timeout=timeout) # Write script to temp file via SFTP (avoids heredoc/pipe conflicts) import hashlib script_hash = hashlib.md5(script.encode()).hexdigest()[:8] tmp_script = f"/tmp/_amnz_script_{script_hash}.sh" self.upload_file(script, tmp_script) # Run with sudo (password via stdin, never on the command line) if self.password: return self.run_command( f"sudo -S -p '' bash {tmp_script}; rm -f {tmp_script}", timeout=timeout, stdin_input=self.password + '\n') return self.run_command(f"sudo bash {tmp_script}; rm -f {tmp_script}", timeout=timeout) def run_script(self, script, timeout=120): """Execute a multi-line script on remote server.""" return self.run_command(script, timeout=timeout) def upload_file(self, content, remote_path): """Upload text content to a remote file via SFTP.""" with self._exec_lock: return self._upload_file_locked(content, remote_path) def _upload_file_locked(self, content, remote_path): self.ensure_connected() # Normalize line endings (Windows CRLF -> Unix LF) content = content.replace('\r\n', '\n') sftp = self.client.open_sftp() try: with sftp.file(remote_path, 'w') as f: f.write(content) finally: sftp.close() def upload_file_sudo(self, content, remote_path): """ Upload text content to a remote file that requires root access. Uses SFTP to write to /tmp, then sudo mv to the target path. Also normalizes line endings to Unix-style (LF). """ self.ensure_connected() # Normalize line endings (Windows CRLF -> Unix LF) content = content.replace('\r\n', '\n') # Write to temp file via SFTP (no sudo needed for /tmp) import hashlib tmp_name = f"/tmp/_amnz_{hashlib.md5(remote_path.encode()).hexdigest()[:8]}" self.upload_file(content, tmp_name) # Move to target with sudo self.run_sudo_command(f"mv {tmp_name} {remote_path}") self.run_sudo_command(f"chmod 644 {remote_path}") return True def download_file(self, remote_path): """Download text content from a remote file.""" with self._exec_lock: return self._download_file_locked(remote_path) def _download_file_locked(self, remote_path): self.ensure_connected() sftp = self.client.open_sftp() try: with sftp.file(remote_path, 'r') as f: return f.read().decode('utf-8', errors='replace') finally: sftp.close() def file_exists(self, remote_path): """Check if a remote file exists.""" with self._exec_lock: return self._file_exists_locked(remote_path) def _file_exists_locked(self, remote_path): self.ensure_connected() sftp = self.client.open_sftp() try: sftp.stat(remote_path) return True except FileNotFoundError: return False finally: sftp.close() def test_connection(self): """Test SSH connection and return server info.""" out, err, code = self.run_command("uname -sr && cat /etc/os-release 2>/dev/null | head -2") return out def write_file(self, remote_path, content): """Write content to a remote file with sudo.""" return self.upload_file_sudo(content, remote_path) def __enter__(self): self.connect() return self def __exit__(self, *args): self.disconnect()