Zhengshou Lai
c1df4098b3
- Move all source files from mytoolkit/ to bin/ - Update entry point and build config in pyproject.toml - Add pillow and pypdf dependencies - Update README and .gitignore paths - Remove unused subprocess import in pdf.py - Clean up duplicate imports in pdf merge command
112 lines
3.1 KiB
Python
112 lines
3.1 KiB
Python
"""SSH connection utilities."""
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import click
|
|
|
|
from bin.config import config
|
|
from bin.utils import handle_errors
|
|
|
|
|
|
def _get_hosts() -> dict:
|
|
"""Get SSH hosts from config."""
|
|
hosts = {}
|
|
for key in config.get_all().keys():
|
|
if key.startswith("ssh_") and not key.endswith("_port") and not key.endswith("_key"):
|
|
name = key[4:] # Remove 'ssh_' prefix
|
|
host = config.get(f"ssh_{name}")
|
|
port = config.get(f"ssh_{name}_port") or "22"
|
|
key_file = config.get(f"ssh_{name}_key")
|
|
if host:
|
|
hosts[name] = {"host": host, "port": int(port)}
|
|
if key_file:
|
|
hosts[name]["key"] = key_file
|
|
return hosts
|
|
|
|
|
|
def _build_ssh_base_cmd(host_config: dict) -> list[str]:
|
|
"""Build base SSH command with host and port."""
|
|
cmd = ["ssh", "-p", str(host_config["port"])]
|
|
|
|
if "key" in host_config:
|
|
key_path = Path(host_config["key"]).expanduser()
|
|
cmd.extend(["-i", str(key_path)])
|
|
|
|
cmd.append(host_config["host"])
|
|
return cmd
|
|
|
|
|
|
@click.group(name="ssh")
|
|
def ssh_cmd():
|
|
"""SSH connection utilities."""
|
|
pass
|
|
|
|
|
|
@ssh_cmd.command("connect")
|
|
@click.argument("name")
|
|
@handle_errors
|
|
def connect(name):
|
|
"""Connect to a configured host."""
|
|
hosts = _get_hosts()
|
|
if name not in hosts:
|
|
click.echo(f"Unknown host: {name}", err=True)
|
|
click.echo(f"Available: {', '.join(hosts.keys())}", err=True)
|
|
raise click.Exit(1)
|
|
|
|
cmd = _build_ssh_base_cmd(hosts[name])
|
|
click.secho(f"Connecting to {name}...", fg="cyan")
|
|
subprocess.run(cmd)
|
|
|
|
|
|
@ssh_cmd.command("tunnel")
|
|
@click.argument("name")
|
|
@click.option("--local-port", "-l", default=11111, help="Local port")
|
|
@click.option("--remote-port", "-r", default=11111, help="Remote port")
|
|
@handle_errors
|
|
def tunnel(name, local_port, remote_port):
|
|
"""Create SSH tunnel with port forwarding."""
|
|
hosts = _get_hosts()
|
|
if name not in hosts:
|
|
click.echo(f"Unknown host: {name}", err=True)
|
|
raise click.Exit(1)
|
|
|
|
host_config = hosts[name]
|
|
cmd = ["ssh", "-p", str(host_config["port"]), "-L", f"{local_port}:localhost:{remote_port}"]
|
|
|
|
if "key" in host_config:
|
|
key_path = Path(host_config["key"]).expanduser()
|
|
cmd.extend(["-i", str(key_path)])
|
|
|
|
cmd.append(host_config["host"])
|
|
|
|
click.secho(f"Creating tunnel {local_port} -> {remote_port} on {name}...", fg="cyan")
|
|
click.secho(f"Command: {' '.join(cmd)}", fg="dim")
|
|
subprocess.run(cmd)
|
|
|
|
|
|
@ssh_cmd.command("list")
|
|
def list_hosts():
|
|
"""List configured SSH hosts."""
|
|
hosts = _get_hosts()
|
|
click.secho("Configured hosts:", fg="cyan")
|
|
for name, host_config in sorted(hosts.items()):
|
|
click.echo(f" {name:12} {host_config['host']}:{host_config['port']}")
|
|
|
|
|
|
@ssh_cmd.command("cmd")
|
|
@click.argument("name")
|
|
@click.argument("command")
|
|
@handle_errors
|
|
def run_cmd(name, command):
|
|
"""Run a command on remote host."""
|
|
hosts = _get_hosts()
|
|
if name not in hosts:
|
|
click.echo(f"Unknown host: {name}", err=True)
|
|
raise click.Exit(1)
|
|
|
|
cmd = _build_ssh_base_cmd(hosts[name])
|
|
cmd.append(command)
|
|
|
|
subprocess.run(cmd)
|