converted into multifile for ease
This commit is contained in:
parent
bfdcf11212
commit
2a7cb7dcb7
9 changed files with 148 additions and 176 deletions
1
.gitignore
vendored
Normal file
1
.gitignore
vendored
Normal file
|
@ -0,0 +1 @@
|
|||
**/__pycache__/
|
0
__init__.py
Normal file
0
__init__.py
Normal file
7
__main__.py
Normal file
7
__main__.py
Normal file
|
@ -0,0 +1,7 @@
|
|||
# ssh_manager/__main__.py
|
||||
|
||||
import sys
|
||||
from .cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
24
add_host.py
24
add_host.py
|
@ -1,24 +1,10 @@
|
|||
# ssh_manager/add_host.py
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from .utils import print_error, print_warning, print_info
|
||||
|
||||
class Colors:
|
||||
GREEN = "\033[0;32m"
|
||||
RED = "\033[0;31m"
|
||||
YELLOW = "\033[1;33m"
|
||||
CYAN = "\033[0;36m"
|
||||
BOLD = "\033[1m"
|
||||
RESET = "\033[0m"
|
||||
|
||||
def print_error(message):
|
||||
print(f"{Colors.RED}{Colors.BOLD}[✖] {Colors.RESET}{message}")
|
||||
|
||||
def print_warning(message):
|
||||
print(f"{Colors.YELLOW}{Colors.BOLD}[⚠] {Colors.RESET}{message}")
|
||||
|
||||
def print_info(message):
|
||||
print(f"{Colors.GREEN}{Colors.BOLD}[✔] {Colors.RESET}{message}")
|
||||
|
||||
def add_host(CONF_DIR):
|
||||
def add_host(conf_dir):
|
||||
"""
|
||||
Interactive prompt to create a new SSH host in ~/.ssh/conf/<label>/config.
|
||||
Offers to generate a new SSH key pair (ed25519) quietly (-q),
|
||||
|
@ -40,7 +26,7 @@ def add_host(CONF_DIR):
|
|||
port = input("Enter SSH port (default: 22): ").strip() or "22"
|
||||
|
||||
# Create subdirectory: ~/.ssh/conf/<label>
|
||||
host_dir = os.path.join(CONF_DIR, host_label)
|
||||
host_dir = os.path.join(conf_dir, host_label)
|
||||
if os.path.exists(host_dir):
|
||||
print_warning(f"Directory {host_dir} already exists; continuing anyway.")
|
||||
else:
|
||||
|
|
57
cli.py
Normal file
57
cli.py
Normal file
|
@ -0,0 +1,57 @@
|
|||
# ssh_manager/cli.py
|
||||
|
||||
import os
|
||||
import glob
|
||||
import asyncio
|
||||
from .utils import print_info, print_error, print_warning, Colors
|
||||
from .config import SSH_DIR, CONF_DIR, SOCKET_DIR, MAIN_CONFIG, DEFAULT_CONFIG_CONTENT
|
||||
from .add_host import add_host
|
||||
from .edit_host import edit_host
|
||||
from .list_hosts import list_hosts
|
||||
|
||||
def ensure_ssh_setup():
|
||||
# Make ~/.ssh if missing
|
||||
if not os.path.isdir(SSH_DIR):
|
||||
os.makedirs(SSH_DIR, mode=0o700, exist_ok=True)
|
||||
print_info(f"Created directory: {SSH_DIR}")
|
||||
|
||||
# Make ~/.ssh/conf if missing
|
||||
if not os.path.isdir(CONF_DIR):
|
||||
os.makedirs(CONF_DIR, mode=0o700, exist_ok=True)
|
||||
print_info(f"Created directory: {CONF_DIR}")
|
||||
|
||||
# Make ~/.ssh/s if missing
|
||||
if not os.path.isdir(SOCKET_DIR):
|
||||
os.makedirs(SOCKET_DIR, mode=0o700, exist_ok=True)
|
||||
print_info(f"Created directory: {SOCKET_DIR}")
|
||||
|
||||
# Create ~/.ssh/config if not present
|
||||
if not os.path.isfile(MAIN_CONFIG):
|
||||
with open(MAIN_CONFIG, "w") as f:
|
||||
f.write(DEFAULT_CONFIG_CONTENT)
|
||||
print_info(f"Created default SSH config at: {MAIN_CONFIG}")
|
||||
|
||||
def main():
|
||||
ensure_ssh_setup()
|
||||
|
||||
while True:
|
||||
print("\n" + f"{Colors.CYAN}{Colors.BOLD}SSH Config Manager Menu{Colors.RESET}")
|
||||
print("1. List Hosts")
|
||||
print("2. Add a Host")
|
||||
print("3. Edit a Host")
|
||||
print("4. Exit")
|
||||
|
||||
choice = input("Select an option (1-4): ").strip()
|
||||
if choice == '1':
|
||||
asyncio.run(list_hosts(CONF_DIR))
|
||||
elif choice == '2':
|
||||
add_host(CONF_DIR)
|
||||
elif choice == '3':
|
||||
edit_host(CONF_DIR)
|
||||
elif choice == '4':
|
||||
print_info("Exiting...")
|
||||
break
|
||||
else:
|
||||
print_error("Invalid choice. Please select 1, 2, 3, or 4.")
|
||||
|
||||
return 0
|
32
config.py
Normal file
32
config.py
Normal file
|
@ -0,0 +1,32 @@
|
|||
# ssh_manager/config.py
|
||||
|
||||
import os
|
||||
|
||||
# Paths
|
||||
SSH_DIR = os.path.expanduser("~/.ssh")
|
||||
CONF_DIR = os.path.join(SSH_DIR, "conf")
|
||||
SOCKET_DIR = os.path.join(SSH_DIR, "s")
|
||||
MAIN_CONFIG = os.path.join(SSH_DIR, "config")
|
||||
|
||||
# Default SSH config content if ~/.ssh/config is missing
|
||||
DEFAULT_CONFIG_CONTENT = """###
|
||||
#Local ssh
|
||||
###
|
||||
|
||||
Include conf/*/config
|
||||
|
||||
###
|
||||
#Catch all ssh config
|
||||
###
|
||||
|
||||
Host *
|
||||
UserKnownHostsFile /dev/null
|
||||
StrictHostKeyChecking no
|
||||
ServerAliveInterval 60
|
||||
ConnectTimeout 60
|
||||
AddKeysToAgent yes
|
||||
EscapeChar `
|
||||
ControlMaster auto
|
||||
ControlPersist 72000
|
||||
ControlPath ~/.ssh/s/%C
|
||||
"""
|
46
edit_host.py
46
edit_host.py
|
@ -1,27 +1,12 @@
|
|||
# ssh_manager/edit_host.py
|
||||
|
||||
import os
|
||||
from collections import OrderedDict
|
||||
|
||||
class Colors:
|
||||
GREEN = "\033[0;32m"
|
||||
RED = "\033[0;31m"
|
||||
YELLOW = "\033[1;33m"
|
||||
CYAN = "\033[0;36m"
|
||||
BOLD = "\033[1m"
|
||||
RESET = "\033[0m"
|
||||
|
||||
def print_error(message):
|
||||
print(f"{Colors.RED}{Colors.BOLD}[✖] {Colors.RESET}{message}")
|
||||
|
||||
def print_warning(message):
|
||||
print(f"{Colors.YELLOW}{Colors.BOLD}[⚠] {Colors.RESET}{message}")
|
||||
|
||||
def print_info(message):
|
||||
print(f"{Colors.GREEN}{Colors.BOLD}[✔] {Colors.RESET}{message}")
|
||||
from .utils import print_error, print_warning, print_info
|
||||
|
||||
def load_config_file(file_path):
|
||||
"""
|
||||
Parse a single SSH config file and return a list of host blocks.
|
||||
Each block is an OrderedDict with keys like 'Host', 'HostName', etc.
|
||||
Parse the given config file into a list of host blocks (OrderedDict).
|
||||
"""
|
||||
blocks = []
|
||||
host_data = None
|
||||
|
@ -35,60 +20,46 @@ def load_config_file(file_path):
|
|||
|
||||
for line in lines:
|
||||
stripped_line = line.strip()
|
||||
# Skip empty lines and comments
|
||||
if not stripped_line or stripped_line.startswith('#'):
|
||||
continue
|
||||
|
||||
# Start of a new Host block
|
||||
if stripped_line.lower().startswith('host '):
|
||||
host_labels = stripped_line.split()[1:]
|
||||
# Pick the first label that isn't a wildcard
|
||||
for label in host_labels:
|
||||
if '*' not in label:
|
||||
# If we already have a host_data in progress, close it out
|
||||
if host_data:
|
||||
blocks.append(host_data)
|
||||
host_data = OrderedDict({'Host': label})
|
||||
break
|
||||
elif host_data:
|
||||
# Split on the first whitespace into key/value
|
||||
if ' ' in stripped_line:
|
||||
key, value = stripped_line.split(None, 1)
|
||||
host_data[key] = value.strip()
|
||||
|
||||
# Add the last block if it exists
|
||||
if host_data:
|
||||
blocks.append(host_data)
|
||||
|
||||
return blocks
|
||||
|
||||
def edit_host(CONF_DIR):
|
||||
def edit_host(conf_dir):
|
||||
"""
|
||||
Let the user update fields for an existing host.
|
||||
1) Ask which host label to edit
|
||||
2) Locate its subdirectory + config
|
||||
3) Update (HostName, User, Port, IdentityFile) as needed
|
||||
4) Rewrite the config
|
||||
Let the user update fields for an existing host in ~/.ssh/conf/<label>/config.
|
||||
"""
|
||||
host_label = input("Enter the Host label to edit: ").strip()
|
||||
if not host_label:
|
||||
print_error("Host label cannot be empty.")
|
||||
return
|
||||
|
||||
host_dir = os.path.join(CONF_DIR, host_label)
|
||||
host_dir = os.path.join(conf_dir, host_label)
|
||||
config_path = os.path.join(host_dir, "config")
|
||||
if not os.path.isfile(config_path):
|
||||
print_warning(f"No config file found at {config_path}; cannot edit this host.")
|
||||
return
|
||||
|
||||
# Load the config file and look for the relevant host block
|
||||
blocks = load_config_file(config_path)
|
||||
if not blocks:
|
||||
print_warning(f"No valid Host blocks found in {config_path}")
|
||||
return
|
||||
|
||||
# We'll assume there's only one block in each config
|
||||
# (the "Host <label>"). If multiple blocks exist, adapt accordingly.
|
||||
target_block = None
|
||||
for b in blocks:
|
||||
if b.get("Host") == host_label:
|
||||
|
@ -110,13 +81,11 @@ def edit_host(CONF_DIR):
|
|||
new_port = input(f"Enter new Port [{old_port}]: ").strip()
|
||||
new_ident = input(f"Enter new IdentityFile [{old_identity}]: ").strip()
|
||||
|
||||
# If user leaves it blank, keep the old value
|
||||
final_hostname = new_hostname if new_hostname else old_hostname
|
||||
final_user = new_user if new_user else old_user
|
||||
final_port = new_port if new_port else old_port
|
||||
final_ident = new_ident if new_ident else old_identity
|
||||
|
||||
# Rebuild the config lines
|
||||
new_config_lines = [
|
||||
f"Host {host_label}",
|
||||
f" HostName {final_hostname}",
|
||||
|
@ -126,7 +95,6 @@ def edit_host(CONF_DIR):
|
|||
if final_ident:
|
||||
new_config_lines.append(f" IdentityFile {final_ident}")
|
||||
|
||||
# Overwrite the file
|
||||
try:
|
||||
with open(config_path, "w") as f:
|
||||
for line in new_config_lines:
|
||||
|
|
|
@ -1,71 +1,28 @@
|
|||
#!/usr/bin/env python3
|
||||
# ssh_manager/list_hosts.py
|
||||
|
||||
import os
|
||||
import sys
|
||||
import glob
|
||||
import socket
|
||||
import asyncio
|
||||
from collections import OrderedDict
|
||||
from tabulate import tabulate
|
||||
from collections import OrderedDict
|
||||
|
||||
# Import from add_host.py
|
||||
from add_host import add_host, print_info, print_error, print_warning, Colors
|
||||
# Import the new edit_host function from edit_host.py
|
||||
from edit_host import edit_host as edit_existing_host
|
||||
from .utils import print_warning, print_error, Colors
|
||||
|
||||
# ------------------ Configuration ------------------
|
||||
SSH_DIR = os.path.expanduser("~/.ssh")
|
||||
CONF_DIR = os.path.join(SSH_DIR, "conf")
|
||||
MAIN_CONFIG = os.path.join(SSH_DIR, "config")
|
||||
SOCKET_DIR = os.path.join(SSH_DIR, "s") # Additional directory for ControlPath
|
||||
|
||||
# Default content to use if ~/.ssh/config doesn't exist
|
||||
DEFAULT_CONFIG_CONTENT = """###
|
||||
#Local ssh
|
||||
###
|
||||
|
||||
Include conf/*/config
|
||||
|
||||
###
|
||||
#Catch all ssh config
|
||||
###
|
||||
|
||||
Host *
|
||||
UserKnownHostsFile /dev/null
|
||||
StrictHostKeyChecking no
|
||||
ServerAliveInterval 60
|
||||
ConnectTimeout 60
|
||||
AddKeysToAgent yes
|
||||
EscapeChar `
|
||||
ControlMaster auto
|
||||
ControlPersist 72000
|
||||
ControlPath ~/.ssh/s/%C
|
||||
"""
|
||||
|
||||
def ensure_ssh_setup():
|
||||
async def check_ssh_port(ip_address, port):
|
||||
"""
|
||||
Ensure ~/.ssh, ~/.ssh/conf, and ~/.ssh/s exist, and if ~/.ssh/config does not exist,
|
||||
create it with DEFAULT_CONFIG_CONTENT.
|
||||
Attempt to open an SSH connection to see if the port is open.
|
||||
Returns True if successful, False otherwise.
|
||||
"""
|
||||
# 1) Create ~/.ssh if it doesn't exist
|
||||
if not os.path.isdir(SSH_DIR):
|
||||
os.makedirs(SSH_DIR, mode=0o700, exist_ok=True)
|
||||
print_info(f"Created directory: {SSH_DIR}")
|
||||
|
||||
# 2) Create ~/.ssh/conf if it doesn't exist
|
||||
if not os.path.isdir(CONF_DIR):
|
||||
os.makedirs(CONF_DIR, mode=0o700, exist_ok=True)
|
||||
print_info(f"Created directory: {CONF_DIR}")
|
||||
|
||||
# 3) Create ~/.ssh/s if it doesn't exist
|
||||
if not os.path.isdir(SOCKET_DIR):
|
||||
os.makedirs(SOCKET_DIR, mode=0o700, exist_ok=True)
|
||||
print_info(f"Created directory: {SOCKET_DIR}")
|
||||
|
||||
# 4) Create ~/.ssh/config if it doesn't exist
|
||||
if not os.path.isfile(MAIN_CONFIG):
|
||||
with open(MAIN_CONFIG, "w") as f:
|
||||
f.write(DEFAULT_CONFIG_CONTENT)
|
||||
print_info(f"Created default SSH config at: {MAIN_CONFIG}")
|
||||
try:
|
||||
reader, writer = await asyncio.wait_for(
|
||||
asyncio.open_connection(ip_address, port), timeout=1
|
||||
)
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
|
||||
def load_config_file(file_path):
|
||||
"""
|
||||
|
@ -84,11 +41,9 @@ def load_config_file(file_path):
|
|||
|
||||
for line in lines:
|
||||
stripped_line = line.strip()
|
||||
# Skip empty lines and comments
|
||||
if not stripped_line or stripped_line.startswith('#'):
|
||||
continue
|
||||
|
||||
# Start of a new Host block
|
||||
if stripped_line.lower().startswith('host '):
|
||||
host_labels = stripped_line.split()[1:]
|
||||
for label in host_labels:
|
||||
|
@ -98,35 +53,18 @@ def load_config_file(file_path):
|
|||
host_data = OrderedDict({'Host': label})
|
||||
break
|
||||
elif host_data:
|
||||
# Split on the first whitespace into key/value
|
||||
if ' ' in stripped_line:
|
||||
key, value = stripped_line.split(None, 1)
|
||||
host_data[key] = value.strip()
|
||||
|
||||
if host_data:
|
||||
blocks.append(host_data)
|
||||
|
||||
return blocks
|
||||
|
||||
async def check_ssh_port(ip_address, port):
|
||||
"""
|
||||
Attempt to open an SSH connection to see if the port is open.
|
||||
Returns True if successful, False otherwise.
|
||||
"""
|
||||
try:
|
||||
reader, writer = await asyncio.wait_for(
|
||||
asyncio.open_connection(ip_address, port), timeout=1
|
||||
)
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
|
||||
async def check_host(host):
|
||||
"""
|
||||
Given a host block (Host, HostName, User, Port, IdentityFile, etc.),
|
||||
resolve the IP, check SSH port, etc. Return a row for tabulate.
|
||||
Given a host block, resolve IP, check SSH port, identity file existence, etc.
|
||||
Return a row for tabulate.
|
||||
"""
|
||||
host_label = host.get('Host', 'N/A')
|
||||
hostname = host.get('HostName', 'N/A')
|
||||
|
@ -134,7 +72,7 @@ async def check_host(host):
|
|||
port = int(host.get('Port', '22'))
|
||||
identity_file = host.get('IdentityFile', 'N/A')
|
||||
|
||||
# Check if identity file exists
|
||||
# Identity file check
|
||||
if identity_file != 'N/A':
|
||||
expanded_identity = os.path.expanduser(identity_file)
|
||||
identity_exists = os.path.isfile(expanded_identity)
|
||||
|
@ -175,12 +113,12 @@ async def check_host(host):
|
|||
identity_display
|
||||
]
|
||||
|
||||
async def list_hosts():
|
||||
async def list_hosts(conf_dir):
|
||||
"""
|
||||
List out all hosts found in ~/.ssh/conf/*/config, showing connectivity details.
|
||||
If no hosts are found, print an empty table or a warning message.
|
||||
"""
|
||||
pattern = os.path.join(CONF_DIR, "*", "config")
|
||||
pattern = os.path.join(conf_dir, "*", "config")
|
||||
conf_files = sorted(glob.glob(pattern))
|
||||
|
||||
all_host_blocks = []
|
||||
|
@ -201,38 +139,3 @@ async def list_hosts():
|
|||
table = [[idx + 1] + row for idx, row in enumerate(results)]
|
||||
print("\nSSH Conf Subdirectory Host List")
|
||||
print(tabulate(table, headers=headers, tablefmt="grid"))
|
||||
|
||||
def interactive_menu():
|
||||
"""
|
||||
A simple interactive menu:
|
||||
1. List Hosts
|
||||
2. Add a Host
|
||||
3. Edit a Host
|
||||
4. Exit
|
||||
"""
|
||||
while True:
|
||||
print("\n" + f"{Colors.CYAN}{Colors.BOLD}SSH Config Manager Menu{Colors.RESET}")
|
||||
print("1. List Hosts")
|
||||
print("2. Add a Host")
|
||||
print("3. Edit a Host")
|
||||
print("4. Exit")
|
||||
|
||||
choice = input("Select an option (1-4): ").strip()
|
||||
if choice == '1':
|
||||
asyncio.run(list_hosts())
|
||||
elif choice == '2':
|
||||
add_host(CONF_DIR)
|
||||
elif choice == '3':
|
||||
edit_existing_host(CONF_DIR)
|
||||
elif choice == '4':
|
||||
print_info("Exiting...")
|
||||
break
|
||||
else:
|
||||
print_error("Invalid choice. Please select 1, 2, 3, or 4.")
|
||||
|
||||
def main():
|
||||
ensure_ssh_setup()
|
||||
interactive_menu()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
18
utils.py
Normal file
18
utils.py
Normal file
|
@ -0,0 +1,18 @@
|
|||
# ssh_manager/utils.py
|
||||
|
||||
class Colors:
|
||||
GREEN = "\033[0;32m"
|
||||
RED = "\033[0;31m"
|
||||
YELLOW = "\033[1;33m"
|
||||
CYAN = "\033[0;36m"
|
||||
BOLD = "\033[1m"
|
||||
RESET = "\033[0m"
|
||||
|
||||
def print_error(message):
|
||||
print(f"{Colors.RED}{Colors.BOLD}[✖] {Colors.RESET}{message}")
|
||||
|
||||
def print_warning(message):
|
||||
print(f"{Colors.YELLOW}{Colors.BOLD}[⚠] {Colors.RESET}{message}")
|
||||
|
||||
def print_info(message):
|
||||
print(f"{Colors.GREEN}{Colors.BOLD}[✔] {Colors.RESET}{message}")
|
Loading…
Add table
Add a link
Reference in a new issue