added additional functions

This commit is contained in:
Arctic 2025-03-07 06:08:30 -06:00
parent 1e55abf811
commit 67a5534246
8 changed files with 654 additions and 107 deletions

View file

@ -1,85 +1,166 @@
# ssh_manager/edit_host.py
import os
import asyncio
from collections import OrderedDict
from .utils import print_error, print_warning, print_info
from .utils import print_error, print_warning, print_info, safe_input
from .list_hosts import list_hosts, load_config_file, check_ssh_port
def load_config_file(file_path):
async def get_all_host_blocks(conf_dir):
"""
Parse the given config file into a list of host blocks (OrderedDict).
Similar to list_hosts, but returns the list of host blocks + a table of results.
We'll build a table ourselves so we can map row numbers to actual host labels.
"""
blocks = []
host_data = None
import glob
import socket
try:
with open(file_path, 'r') as f:
lines = f.readlines()
except Exception as e:
print_error(f"Error reading SSH config file {file_path}: {e}")
return blocks
pattern = os.path.join(conf_dir, "*", "config")
conf_files = sorted(glob.glob(pattern))
for line in lines:
stripped_line = line.strip()
if not stripped_line or stripped_line.startswith('#'):
continue
all_blocks = []
for conf_file in conf_files:
blocks = load_config_file(conf_file)
all_blocks.extend(blocks)
if stripped_line.lower().startswith('host '):
host_labels = stripped_line.split()[1:]
for label in host_labels:
if '*' not in label:
if host_data:
blocks.append(host_data)
host_data = OrderedDict({'Host': label})
break
elif host_data:
if ' ' in stripped_line:
key, value = stripped_line.split(None, 1)
host_data[key] = value.strip()
# If no blocks found, return empty
if not all_blocks:
return []
if host_data:
blocks.append(host_data)
return blocks
# We want to do a partial version of check_host to get row data
# so we can display the table right here and keep track of each blocks host label.
# But let's do it similarly to list_hosts:
table_rows = []
for idx, b in enumerate(all_blocks, start=1):
host_label = b.get("Host", "N/A")
hostname = b.get("HostName", "N/A")
user = b.get("User", "N/A")
port = int(b.get("Port", "22"))
identity_file = b.get("IdentityFile", "N/A")
# Identity check
if identity_file != "N/A":
expanded_identity = os.path.expanduser(identity_file)
identity_exists = os.path.isfile(expanded_identity)
else:
identity_exists = False
# IP resolution
try:
ip_address = socket.gethostbyname(hostname)
except socket.error:
ip_address = None
# Port check
if ip_address:
port_open = await asyncio.wait_for(check_ssh_port(ip_address, port), timeout=1)
else:
port_open = False
# Colors for display (optional, or we can keep it simple):
ip_display = f"\033[0;32m{ip_address}\033[0m" if ip_address else "\033[0;31mN/A\033[0m"
port_display = f"\033[0;32m{port}\033[0m" if port_open else f"\033[0;31m{port}\033[0m"
identity_disp= f"\033[0;32m{identity_file}\033[0m" if identity_exists else f"\033[0;31m{identity_file}\033[0m"
row = [
idx,
host_label,
user,
port_display,
hostname,
ip_display,
identity_disp
]
table_rows.append(row)
# Print the table
from tabulate import tabulate
headers = ["No.", "Host", "User", "Port", "HostName", "IP Address", "IdentityFile"]
print("\nSSH Conf Subdirectory Host List")
print(tabulate(table_rows, headers=headers, tablefmt="grid"))
return all_blocks
def edit_host(conf_dir):
"""
Let the user update fields for an existing host in ~/.ssh/conf/<label>/config.
The user may type either the row number OR the actual host label.
"""
host_label = input("Enter the Host label to edit: ").strip()
if not host_label:
print_error("Host label cannot be empty.")
# 1) Gather + display the current host list
print_info("Here is the current list of hosts:\n")
all_blocks = asyncio.run(get_all_host_blocks(conf_dir))
if not all_blocks:
print_warning("No hosts found to edit.")
return
# 2) Prompt for which host to edit (by label or row number)
choice = safe_input("Enter the row number or the Host label to edit: ")
if choice is None:
return # user canceled (Ctrl+C)
choice = choice.strip()
if not choice:
print_error("Host label or row number cannot be empty.")
return
# Check if user typed a digit -> row number
target_block = None
if choice.isdigit():
row_idx = int(choice)
# Validate index
if row_idx < 1 or row_idx > len(all_blocks):
print_warning(f"Invalid row number {row_idx}.")
return
target_block = all_blocks[row_idx - 1] # zero-based
else:
# The user typed a host label
# We must search all_blocks for a matching Host
for b in all_blocks:
if b.get("Host") == choice:
target_block = b
break
if not target_block:
print_warning(f"No matching host label '{choice}' found.")
return
# Now we have a target_block with existing data
host_label = target_block.get("Host", "")
if not host_label:
print_warning("This host block has no label. Cannot edit.")
return
# Derive the config path
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
blocks = load_config_file(config_path)
if not blocks:
print_warning(f"No valid Host blocks found in {config_path}")
return
target_block = None
for b in blocks:
if b.get("Host") == host_label:
target_block = b
break
if not target_block:
print_warning(f"No matching Host '{host_label}' found in {config_path}")
return
old_hostname = target_block.get("HostName", "")
old_user = target_block.get("User", "")
old_port = target_block.get("Port", "22")
old_identity = target_block.get("IdentityFile", "")
print_info("Leave a field blank to keep its current value.")
new_hostname = input(f"Enter new HostName [{old_hostname}]: ").strip()
new_user = input(f"Enter new User [{old_user}]: ").strip()
new_port = input(f"Enter new Port [{old_port}]: ").strip()
new_ident = input(f"Enter new IdentityFile [{old_identity}]: ").strip()
new_hostname = safe_input(f"Enter new HostName [{old_hostname}]: ")
if new_hostname is None:
return
new_hostname = new_hostname.strip()
new_user = safe_input(f"Enter new User [{old_user}]: ")
if new_user is None:
return
new_user = new_user.strip()
new_port = safe_input(f"Enter new Port [{old_port}]: ")
if new_port is None:
return
new_port = new_port.strip()
new_ident = safe_input(f"Enter new IdentityFile [{old_identity}]: ")
if new_ident is None:
return
new_ident = new_ident.strip()
final_hostname = new_hostname if new_hostname else old_hostname
final_user = new_user if new_user else old_user