v1.8.65: Linux→Windows auto-sanitize commands for Windows SSH servers
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
chrome-storm-c442
@@ -246,7 +246,10 @@ class SSHClientWrapper:
|
||||
sftp.put(local_path, remote_path, callback=progress_cb)
|
||||
else:
|
||||
sftp.put(local_path, remote_path)
|
||||
try:
|
||||
sftp.chmod(remote_path, 0o664)
|
||||
except OSError:
|
||||
pass # Windows OpenSSH doesn't support chmod
|
||||
sftp.close()
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
279
tools/ssh.py
279
tools/ssh.py
@@ -149,9 +149,13 @@ def run_command(server: dict, command: str, use_sudo: bool = True) -> tuple:
|
||||
client = get_client(server)
|
||||
try:
|
||||
user = server.get("user", "root")
|
||||
need_sudo = use_sudo and user != "root"
|
||||
is_win = _is_windows_server(server)
|
||||
need_sudo = not is_win and use_sudo and user != "root"
|
||||
|
||||
if need_sudo:
|
||||
if is_win:
|
||||
# Windows SSH: translate Linux commands to Windows equivalents
|
||||
full_cmd = _sanitize_windows_command(command)
|
||||
elif need_sudo:
|
||||
# Use sudo -S to read password from stdin
|
||||
# -p '' suppresses the password prompt text
|
||||
full_cmd = f"sudo -S -p '' bash -c {_shell_quote(command)}"
|
||||
@@ -184,10 +188,278 @@ def _shell_quote(s: str) -> str:
|
||||
return "'" + s.replace("'", "'\\''") + "'"
|
||||
|
||||
|
||||
def _is_windows_server(server: dict) -> bool:
|
||||
"""Detect if server is Windows by alias, notes, or type."""
|
||||
stype = server.get("type", "ssh")
|
||||
if stype in ("winrm", "rdp"):
|
||||
return True
|
||||
text = (server.get("alias", "") + " " + server.get("notes", "")).lower()
|
||||
return "windows" in text or "win " in text
|
||||
|
||||
|
||||
# ── Windows command translation ──────────────────────
|
||||
|
||||
# Map of Linux commands → Windows/PowerShell equivalents
|
||||
# Each entry: (regex_pattern, replacement_callable)
|
||||
# Patterns use ^\s* anchoring — matches only at command start, no partial-word risk
|
||||
_WIN_CMD_MAP = [
|
||||
# --- Skip / warn ---
|
||||
(r'^\s*chmod\b.*', lambda m: 'echo [SKIP] chmod not supported on Windows'),
|
||||
(r'^\s*chown\b.*', lambda m: 'echo [SKIP] chown not supported on Windows'),
|
||||
# --- Environment ---
|
||||
(r'^\s*export\s+(\w+)=(.*)', lambda m: f'set {m.group(1)}={m.group(2)}'),
|
||||
(r'^\s*env\s*$', lambda m: 'set'),
|
||||
(r'^\s*printenv\s*$', lambda m: 'set'),
|
||||
(r'^\s*printenv\s+(\w+)', lambda m: f'echo %{m.group(1)}%'),
|
||||
# --- File operations ---
|
||||
(r'^\s*ls\s+-la\s+(.*)', lambda m: f'dir /a "{m.group(1).strip()}"'),
|
||||
(r'^\s*ls\s+-lah?\s+(.*)', lambda m: f'dir /a "{m.group(1).strip()}"'),
|
||||
(r'^\s*ls\s+-[a-zA-Z]*R[a-zA-Z]*\s*(.*)', lambda m: f'dir /s /a "{m.group(1).strip()}"' if m.group(1).strip() else 'dir /s /a'),
|
||||
(r'^\s*ls\s+-[a-zA-Z]+\s+(.*)', lambda m: f'dir "{m.group(1).strip()}"'),
|
||||
(r'^\s*ls\s+-[a-zA-Z]+\s*$', lambda m: 'dir'),
|
||||
(r'^\s*ls\s+(.*)', lambda m: f'dir "{m.group(1).strip()}"'),
|
||||
(r'^\s*ls\s*$', lambda m: 'dir'),
|
||||
(r'^\s*cat\b\s+(.*)', lambda m: f'type {m.group(1).strip()}'),
|
||||
(r'^\s*cp\s+-r\s+(.*?)\s+(.*)', lambda m: f'xcopy /E /I "{m.group(1).strip()}" "{m.group(2).strip()}"'),
|
||||
(r'^\s*cp\s+(.*?)\s+(.*)', lambda m: f'copy "{m.group(1).strip()}" "{m.group(2).strip()}"'),
|
||||
(r'^\s*mv\s+(.*?)\s+(.*)', lambda m: f'move "{m.group(1).strip()}" "{m.group(2).strip()}"'),
|
||||
(r'^\s*rm\s+-rf?\s+(.*)', lambda m: f'rmdir /S /Q "{m.group(1).strip()}"'),
|
||||
(r'^\s*rm\s+-f\s+(.*)', lambda m: f'del /F /Q "{m.group(1).strip()}"'),
|
||||
(r'^\s*rm\s+(.*)', lambda m: f'del /Q "{m.group(1).strip()}"'),
|
||||
(r'^\s*mkdir\s+-p\s+(.*)', lambda m: f'mkdir "{m.group(1).strip()}"'),
|
||||
(r'^\s*touch\s+(.*)', lambda m: f'type nul > "{m.group(1).strip()}"'),
|
||||
(r'^\s*pwd\s*$', lambda m: 'cd'),
|
||||
# --- Text processing (PowerShell) ---
|
||||
(r'^\s*grep\s+-r\s+"?([^"]*)"?\s+(.*)', lambda m: f'powershell -Command "Get-ChildItem -Recurse {m.group(2).strip()} | Select-String -Pattern \'{m.group(1)}\'"'),
|
||||
(r'^\s*grep\s+-i\s+"?([^"]*)"?\s+(.*)', lambda m: f'findstr /I "{m.group(1)}" {m.group(2).strip()}'),
|
||||
(r'^\s*grep\s+"?([^"]*)"?\s+(.*)', lambda m: f'findstr "{m.group(1)}" {m.group(2).strip()}'),
|
||||
(r'^\s*head\s+-n?\s*(\d+)\s+(.*)', lambda m: f'powershell -Command "Get-Content \'{m.group(2).strip()}\' | Select-Object -First {m.group(1)}"'),
|
||||
(r'^\s*head\s+(.*)', lambda m: f'powershell -Command "Get-Content \'{m.group(1).strip()}\' | Select-Object -First 10"'),
|
||||
(r'^\s*tail\s+-n?\s*(\d+)\s+(.*)', lambda m: f'powershell -Command "Get-Content \'{m.group(2).strip()}\' -Tail {m.group(1)}"'),
|
||||
(r'^\s*tail\s+-f\s+(.*)', lambda m: f'powershell -Command "Get-Content \'{m.group(1).strip()}\' -Wait -Tail 20"'),
|
||||
(r'^\s*tail\s+(.*)', lambda m: f'powershell -Command "Get-Content \'{m.group(1).strip()}\' -Tail 10"'),
|
||||
(r'^\s*wc\s+-l\s+(.*)', lambda m: f'powershell -Command "(Get-Content \'{m.group(1).strip()}\' | Measure-Object -Line).Lines"'),
|
||||
# --- Search ---
|
||||
(r'^\s*find\s+(.*?)\s+-name\s+"?([^"]*)"?', lambda m: f'powershell -Command "Get-ChildItem -Path \'{m.group(1).strip()}\' -Recurse -Filter \'{m.group(2)}\'"'),
|
||||
(r'^\s*which\s+(.*)', lambda m: f'where.exe {m.group(1).strip()}'),
|
||||
(r'^\s*whereis\s+(.*)', lambda m: f'where.exe {m.group(1).strip()}'),
|
||||
# --- Process / system ---
|
||||
(r'^\s*ps\s+aux\s*$', lambda m: 'tasklist'),
|
||||
(r'^\s*ps\s+-ef\s*$', lambda m: 'tasklist'),
|
||||
(r'^\s*ps\s*$', lambda m: 'tasklist'),
|
||||
(r'^\s*kill\s+-9\s+(\d+)', lambda m: f'taskkill /F /PID {m.group(1)}'),
|
||||
(r'^\s*kill\s+(\d+)', lambda m: f'taskkill /PID {m.group(1)}'),
|
||||
(r'^\s*top\s*$', lambda m: 'powershell -Command "Get-Process | Sort-Object CPU -Descending | Select-Object -First 20 Name, Id, CPU, @{N=\'Mem(MB)\';E={[math]::Round($_.WS/1MB,1)}}"'),
|
||||
(r'^\s*df\s*', lambda m: 'powershell -Command "Get-PSDrive -PSProvider FileSystem | Select-Object Name, @{N=\'Used(GB)\';E={[math]::Round($_.Used/1GB,1)}}, @{N=\'Free(GB)\';E={[math]::Round($_.Free/1GB,1)}}"'),
|
||||
(r'^\s*free\b.*', lambda m: 'powershell -Command "$os=Get-CimInstance Win32_OperatingSystem; [PSCustomObject]@{TotalMB=[math]::Round($os.TotalVisibleMemorySize/1024); FreeMB=[math]::Round($os.FreePhysicalMemory/1024); UsedMB=[math]::Round(($os.TotalVisibleMemorySize-$os.FreePhysicalMemory)/1024)} | Format-List"'),
|
||||
(r'^\s*du\s+-sh?\s+(.*)', lambda m: f'powershell -Command "(Get-ChildItem -Recurse \'{m.group(1).strip()}\' | Measure-Object -Property Length -Sum).Sum / 1MB | ForEach-Object {{Write-Host ([math]::Round($_,1)) \'MB\'}}"'),
|
||||
(r'^\s*uname\b.*', lambda m: 'powershell -Command "[System.Environment]::OSVersion | Format-List"'),
|
||||
(r'^\s*uptime\s*$', lambda m: 'powershell -Command "$b=(Get-CimInstance Win32_OperatingSystem).LastBootUpTime; $u=(New-TimeSpan $b (Get-Date)); Write-Host \"up $($u.Days)d $($u.Hours)h $($u.Minutes)m, since $b\""'),
|
||||
(r'^\s*whoami\s*$', lambda m: 'whoami'),
|
||||
(r'^\s*hostname\s*$', lambda m: 'hostname'),
|
||||
# --- Service management ---
|
||||
(r'^\s*systemctl\s+status\s+(.*)', lambda m: f'powershell -Command "Get-Service \'{m.group(1).strip()}\' | Format-List"'),
|
||||
(r'^\s*systemctl\s+start\s+(.*)', lambda m: f'powershell -Command "Start-Service \'{m.group(1).strip()}\'"'),
|
||||
(r'^\s*systemctl\s+stop\s+(.*)', lambda m: f'powershell -Command "Stop-Service \'{m.group(1).strip()}\'"'),
|
||||
(r'^\s*systemctl\s+restart\s+(.*)', lambda m: f'powershell -Command "Restart-Service \'{m.group(1).strip()}\'"'),
|
||||
(r'^\s*systemctl\s+list-units\b.*', lambda m: 'powershell -Command "Get-Service | Format-Table Name, Status, DisplayName -AutoSize"'),
|
||||
# --- Networking ---
|
||||
(r'^\s*ifconfig\s*$', lambda m: 'ipconfig'),
|
||||
(r'^\s*ip\s+addr\b.*', lambda m: 'ipconfig /all'),
|
||||
(r'^\s*ip\s+route\b.*', lambda m: 'route print'),
|
||||
(r'^\s*netstat\b(.*)', lambda m: f'netstat{m.group(1)}'),
|
||||
(r'^\s*ss\s+-tulnp\s*$', lambda m: 'netstat -ano'),
|
||||
(r'^\s*curl\s+(.*)', lambda m: f'powershell -Command "Invoke-WebRequest -Uri \'{m.group(1).strip()}\' -UseBasicParsing | Select-Object StatusCode, Content"'),
|
||||
(r'^\s*wget\s+(.*)', lambda m: f'powershell -Command "Invoke-WebRequest -Uri \'{m.group(1).strip()}\' -OutFile ([System.IO.Path]::GetFileName(\'{m.group(1).strip()}\'))"'),
|
||||
]
|
||||
|
||||
|
||||
def _strip_sudo(command: str) -> str:
|
||||
"""Strip sudo and its flags from a command. Handles -u USER, -S, -p '', etc."""
|
||||
# sudo flags that consume the next argument
|
||||
_SUDO_ARG_FLAGS = {'-u', '-g', '-C', '-p', '-r', '-t', '-D'}
|
||||
parts = command.strip().split()
|
||||
if not parts or parts[0] != 'sudo':
|
||||
return command
|
||||
i = 1
|
||||
while i < len(parts):
|
||||
if parts[i].startswith('-'):
|
||||
flag = parts[i]
|
||||
i += 1
|
||||
if flag in _SUDO_ARG_FLAGS and i < len(parts):
|
||||
i += 1 # skip argument
|
||||
else:
|
||||
break
|
||||
return ' '.join(parts[i:]) if i < len(parts) else ''
|
||||
|
||||
|
||||
def _sanitize_windows_command(command: str) -> str:
|
||||
"""Translate Linux commands to Windows/PowerShell equivalents for Windows SSH servers.
|
||||
|
||||
Features:
|
||||
- Translates common Linux commands (ls, cat, grep, ps, df, etc.)
|
||||
- Wraps pipe chains in PowerShell when needed
|
||||
- Forces UTF-8 via chcp 65001
|
||||
- Passes through native Windows/PowerShell commands unchanged
|
||||
- Auto-removes sudo, skips chmod/chown with warning
|
||||
"""
|
||||
import re
|
||||
|
||||
# 0. Empty / whitespace guard
|
||||
if not command or not command.strip():
|
||||
return command
|
||||
|
||||
# 1. Strip sudo early (before any other logic)
|
||||
if re.match(r'^\s*sudo\b', command):
|
||||
command = _strip_sudo(command)
|
||||
if not command:
|
||||
return command
|
||||
|
||||
# 2. Handle && chains — split, translate each, rejoin (before passthrough!)
|
||||
if ' && ' in command:
|
||||
parts = command.split(' && ')
|
||||
translated = [_translate_single_command(p.strip()) for p in parts]
|
||||
joined = ' && '.join(translated)
|
||||
return f"chcp 65001 >nul && {joined}"
|
||||
|
||||
# 3. Handle pipes (before passthrough!)
|
||||
if '|' in command:
|
||||
return _translate_piped_command(command)
|
||||
|
||||
# 4. Passthrough — already Windows/PowerShell native (single commands only)
|
||||
stripped = command.strip()
|
||||
passthrough_prefixes = ('powershell ', 'powershell.exe ', 'pwsh ', 'pwsh.exe ',
|
||||
'cmd /c ', 'cmd.exe /c ', 'chcp ', 'dir ', 'type ',
|
||||
'copy ', 'move ', 'del ', 'mkdir ', 'rmdir ',
|
||||
'tasklist', 'taskkill', 'ipconfig', 'netstat',
|
||||
'systeminfo', 'where.exe', 'findstr ', 'echo ',
|
||||
'set ', 'reg ', 'sc ', 'wmic ',
|
||||
'Get-', 'Set-', 'New-', 'Remove-', 'Start-', 'Stop-',
|
||||
'Restart-', 'Invoke-', 'Select-', 'Write-', 'Out-',
|
||||
'Format-', 'Test-', 'Import-', 'Export-')
|
||||
stripped_lower = stripped.lower()
|
||||
for prefix in passthrough_prefixes:
|
||||
if stripped_lower.startswith(prefix.lower()):
|
||||
return f"chcp 65001 >nul && {command}"
|
||||
|
||||
# 5. Single command translation
|
||||
translated = _translate_single_command(command)
|
||||
return f"chcp 65001 >nul && {translated}"
|
||||
|
||||
|
||||
def _translate_single_command(command: str) -> str:
|
||||
"""Translate a single (non-piped) Linux command to Windows equivalent."""
|
||||
import re
|
||||
for pattern, replacement in _WIN_CMD_MAP:
|
||||
match = re.match(pattern, command.strip(), re.IGNORECASE)
|
||||
if match:
|
||||
try:
|
||||
return replacement(match)
|
||||
except Exception:
|
||||
continue
|
||||
return command # No match — pass through as-is
|
||||
|
||||
|
||||
def _translate_piped_command(command: str) -> str:
|
||||
"""Translate a piped Linux command chain to PowerShell."""
|
||||
import re
|
||||
# If it already contains PowerShell cmdlets, pass through
|
||||
if re.search(r'Get-|Set-|Select-|Where-|ForEach-|Measure-', command):
|
||||
return f"chcp 65001 >nul && {command}"
|
||||
|
||||
# Split by pipe, translate first command, check if we need PowerShell wrapping
|
||||
parts = [p.strip() for p in command.split('|')]
|
||||
|
||||
# Try translating the first segment
|
||||
first_translated = _translate_single_command(parts[0])
|
||||
|
||||
# If the pipe chain mixes cmd and grep/awk/sed — wrap entire thing in PowerShell
|
||||
needs_ps = any(re.match(r'\s*(grep|awk|sed|sort|uniq|wc|head|tail|cut|tr)\b', p, re.IGNORECASE)
|
||||
for p in parts[1:])
|
||||
|
||||
if needs_ps:
|
||||
# Build a PowerShell pipeline
|
||||
ps_parts = [_linux_to_ps_pipe_segment(p.strip()) for p in parts]
|
||||
ps_cmd = ' | '.join(ps_parts)
|
||||
return f'chcp 65001 >nul && powershell -Command "{ps_cmd}"'
|
||||
|
||||
# Otherwise just join translated parts
|
||||
translated_parts = [first_translated] + [_translate_single_command(p) for p in parts[1:]]
|
||||
return f"chcp 65001 >nul && {' | '.join(translated_parts)}"
|
||||
|
||||
|
||||
def _linux_to_ps_pipe_segment(segment: str) -> str:
|
||||
"""Convert a single pipe segment from Linux to PowerShell equivalent."""
|
||||
import re
|
||||
s = segment.strip()
|
||||
# cat → Get-Content
|
||||
m = re.match(r'cat\b\s+(.*)', s, re.IGNORECASE)
|
||||
if m:
|
||||
return f"Get-Content '{m.group(1).strip()}'"
|
||||
# ps aux/ps -ef → Get-Process
|
||||
if re.match(r'ps\s+(aux|-ef)\s*$', s, re.IGNORECASE):
|
||||
return 'Get-Process'
|
||||
# grep → Select-String
|
||||
m = re.match(r'grep\s+(-i\s+)?"?([^"]*)"?', s, re.IGNORECASE)
|
||||
if m:
|
||||
return f"Select-String -Pattern '{m.group(2)}'"
|
||||
# sort
|
||||
if re.match(r'sort\s*$', s, re.IGNORECASE):
|
||||
return 'Sort-Object'
|
||||
# sort -r / sort -n
|
||||
if re.match(r'sort\s+-r', s, re.IGNORECASE):
|
||||
return 'Sort-Object -Descending'
|
||||
if re.match(r'sort\s+-n', s, re.IGNORECASE):
|
||||
return 'Sort-Object {[int]$_}'
|
||||
# uniq
|
||||
if re.match(r'uniq\s*$', s, re.IGNORECASE):
|
||||
return 'Get-Unique'
|
||||
# wc -l
|
||||
if re.match(r'wc\s+-l', s, re.IGNORECASE):
|
||||
return 'Measure-Object -Line'
|
||||
# head -n N
|
||||
m = re.match(r'head\s+(-n\s*)?(\d+)', s, re.IGNORECASE)
|
||||
if m:
|
||||
return f'Select-Object -First {m.group(2)}'
|
||||
if re.match(r'head\s*$', s, re.IGNORECASE):
|
||||
return 'Select-Object -First 10'
|
||||
# tail -n N
|
||||
m = re.match(r'tail\s+(-n\s*)?(\d+)', s, re.IGNORECASE)
|
||||
if m:
|
||||
return f'Select-Object -Last {m.group(2)}'
|
||||
if re.match(r'tail\s*$', s, re.IGNORECASE):
|
||||
return 'Select-Object -Last 10'
|
||||
# awk — basic field extraction
|
||||
m = re.match(r"awk\s+['\"]?\{print\s+\$(\d+)\}['\"]?", s, re.IGNORECASE)
|
||||
if m:
|
||||
idx = int(m.group(1)) - 1
|
||||
return f"ForEach-Object {{ ($_ -split '\\s+')[{idx}] }}"
|
||||
# cut -d'X' -f N
|
||||
m = re.match(r"cut\s+-d['\"]?(.)['\"]?\s+-f(\d+)", s, re.IGNORECASE)
|
||||
if m:
|
||||
idx = int(m.group(2)) - 1
|
||||
return f"ForEach-Object {{ ($_ -split '{m.group(1)}')[{idx}] }}"
|
||||
# sed — basic s/old/new/
|
||||
m = re.match(r"sed\s+['\"]?s/([^/]*)/([^/]*)/?[g]?['\"]?", s, re.IGNORECASE)
|
||||
if m:
|
||||
return f"ForEach-Object {{ $_ -replace '{m.group(1)}', '{m.group(2)}' }}"
|
||||
# tr -d 'X'
|
||||
m = re.match(r"tr\s+-d\s+['\"]?(.+?)['\"]?\s*$", s, re.IGNORECASE)
|
||||
if m:
|
||||
return f"ForEach-Object {{ $_ -replace '[{m.group(1)}]', '' }}"
|
||||
# Fallback — return as-is
|
||||
return s
|
||||
|
||||
|
||||
# ── File transfer ─────────────────────────────────────
|
||||
|
||||
def _normalize_remote_path(remote_path: str) -> str:
|
||||
"""Normalize remote path by detecting and fixing MSYS path conversions."""
|
||||
# Strip leading // that user adds to prevent Git Bash MSYS path conversion
|
||||
# //C:/Temp → /C:/Temp
|
||||
if remote_path.startswith("//") and len(remote_path) > 2 and remote_path[2] != "/":
|
||||
remote_path = remote_path[1:]
|
||||
|
||||
# If the path looks like a Windows path that was converted by MSYS, fix it back
|
||||
if ':' in remote_path and ('Program Files/Git' in remote_path or (len(remote_path) > 3 and remote_path[1] == ':' and remote_path[2] == '/')):
|
||||
# Convert C:/Program Files/Git/tmp/file.txt back to /tmp/file.txt
|
||||
@@ -251,7 +523,10 @@ def upload_file(server: dict, local_path: str, remote_path: str):
|
||||
t0 = time.time()
|
||||
sftp.put(local_path, normalized_remote_path, callback=_progress_cb(file_size))
|
||||
elapsed = time.time() - t0
|
||||
try:
|
||||
sftp.chmod(normalized_remote_path, 0o664)
|
||||
except OSError:
|
||||
pass # Windows OpenSSH doesn't support chmod
|
||||
sftp.close()
|
||||
|
||||
info = f"{_fmt_size(file_size)}, {elapsed:.1f}s"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Version info for ServerManager."""
|
||||
|
||||
__version__ = "1.8.63"
|
||||
__version__ = "1.8.65"
|
||||
__app_name__ = "ServerManager"
|
||||
__author__ = "aibot777"
|
||||
__description__ = "Desktop GUI for managing remote servers"
|
||||
|
||||
Reference in New Issue
Block a user