236 lines
10 KiB
Python
236 lines
10 KiB
Python
import subprocess
|
||
from dialog import Dialog
|
||
import logging
|
||
import os
|
||
|
||
SSH_CONFIG_PATH = "/etc/ssh/sshd_config"
|
||
SSH_AUTH_CONFIG_PATH = "/etc/ssh/sshd_config.d/auth.conf"
|
||
UFW_OPENSSH_CONFIG_PATH = "/etc/ufw/applications.d/openssh-server"
|
||
|
||
|
||
class SSHConfig:
|
||
def __init__(self):
|
||
self.d = Dialog(dialog="dialog")
|
||
self.current_port = self.get_current_ssh_port()
|
||
self.current_auth_method = self.get_current_auth_method()
|
||
self.new_port = None
|
||
self.new_auth_method = None
|
||
|
||
def get_sshd_status(self):
|
||
try:
|
||
result = subprocess.run(
|
||
["systemctl", "is-active", "sshd"], capture_output=True, text=True
|
||
)
|
||
return result.stdout.strip()
|
||
except subprocess.CalledProcessError:
|
||
return "unknown"
|
||
|
||
def toggle_sshd_service(self, action):
|
||
try:
|
||
subprocess.run(["sudo", "systemctl", action, "sshd"], check=True)
|
||
self.d.msgbox(
|
||
f"Служба sshd успешно {'запущена' if action == 'start' else 'остановлена'}."
|
||
)
|
||
except subprocess.CalledProcessError as e:
|
||
logging.error(f"Ошибка при {action} службы sshd: {e}")
|
||
self.d.msgbox(
|
||
f"Ошибка при {'запуске' if action == 'start' else 'остановке'} службы sshd."
|
||
)
|
||
|
||
def get_current_ssh_port(self):
|
||
try:
|
||
result = subprocess.run(
|
||
["grep", "^#*Port", SSH_CONFIG_PATH],
|
||
capture_output=True,
|
||
text=True,
|
||
check=True,
|
||
)
|
||
port = result.stdout.strip().split()[-1]
|
||
return port if port.isdigit() else "22"
|
||
except subprocess.CalledProcessError:
|
||
logging.warning(
|
||
"Не удалось получить порт SSH, используется порт по умолчанию"
|
||
)
|
||
return "22"
|
||
|
||
def change_ssh_port(self):
|
||
port_display = self.new_port if self.new_port else self.current_port
|
||
port_status = " (не применено)" if self.new_port else ""
|
||
code, new_port = self.d.inputbox(
|
||
f"Текущий порт SSH: {port_display}{port_status}\nВведите новый порт SSH:",
|
||
init=port_display,
|
||
)
|
||
if code != self.d.OK:
|
||
return
|
||
try:
|
||
new_port = int(new_port)
|
||
if new_port < 1 or new_port > 65535:
|
||
raise ValueError
|
||
except ValueError:
|
||
self.d.msgbox("Некорректный номер порта. Введите число от 1 до 65535.")
|
||
return
|
||
code = self.d.yesno(f"Вы уверены, что хотите изменить порт SSH на {new_port}?")
|
||
if code != self.d.OK:
|
||
return
|
||
self.new_port = str(new_port)
|
||
|
||
def get_current_auth_method(self):
|
||
try:
|
||
with open(SSH_AUTH_CONFIG_PATH, "r") as f:
|
||
content = f.read()
|
||
password_auth = "yes" if "PasswordAuthentication yes" in content else "no"
|
||
pubkey_auth = "yes" if "PubkeyAuthentication yes" in content else "no"
|
||
|
||
if password_auth == "yes" and pubkey_auth == "yes":
|
||
return "По ключу и паролю"
|
||
elif password_auth == "no" and pubkey_auth == "yes":
|
||
return "Только по ключу"
|
||
elif password_auth == "yes" and pubkey_auth == "no":
|
||
return "Только по паролю"
|
||
else:
|
||
return "По ключу и паролю" # Default value
|
||
except FileNotFoundError:
|
||
logging.warning("Файл конфигурации аутентификации SSH не найден")
|
||
return "По ключу и паролю" # Default value
|
||
except Exception as e:
|
||
logging.warning(f"Не удалось получить метод аутентификации SSH: {e}")
|
||
return "По ключу и паролю" # Default value
|
||
|
||
def change_auth_method(self):
|
||
auth_method_display = (
|
||
self.new_auth_method if self.new_auth_method else self.current_auth_method
|
||
)
|
||
auth_status = " (не применено)" if self.new_auth_method else ""
|
||
auth_choices = [
|
||
(
|
||
"1",
|
||
f"Только по паролю{' (текущий)' if auth_method_display == 'Только по паролю' else ''}",
|
||
),
|
||
(
|
||
"2",
|
||
f"Только по ключу{' (текущий)' if auth_method_display == 'Только по ключу' else ''}",
|
||
),
|
||
(
|
||
"3",
|
||
f"По ключу и паролю{' (текущий)' if auth_method_display == 'По ключу и паролю' else ''}",
|
||
),
|
||
]
|
||
code, tag = self.d.menu(
|
||
f"Текущий метод аутентификации: {auth_method_display}{auth_status}\nВыберите метод аутентификации:",
|
||
choices=auth_choices,
|
||
)
|
||
if code != self.d.OK:
|
||
return
|
||
self.new_auth_method = {
|
||
"1": "Только по паролю",
|
||
"2": "Только по ключу",
|
||
"3": "По ключу и паролю",
|
||
}[tag]
|
||
|
||
def apply_changes(self):
|
||
if self.new_port:
|
||
try:
|
||
subprocess.run(
|
||
[
|
||
"sudo",
|
||
"sed",
|
||
"-i",
|
||
f"s/^#*Port .*/Port {self.new_port}/",
|
||
SSH_CONFIG_PATH,
|
||
],
|
||
check=True,
|
||
)
|
||
subprocess.run(
|
||
[
|
||
"sudo",
|
||
"sed",
|
||
"-i",
|
||
f"s/^ports=.*/ports={self.new_port}\/tcp/",
|
||
UFW_OPENSSH_CONFIG_PATH,
|
||
],
|
||
check=True,
|
||
)
|
||
subprocess.run(["sudo", "ufw", "app", "update", "OpenSSH"], check=True)
|
||
subprocess.run(
|
||
["sudo", "ufw", "delete", "allow", "OpenSSH"], check=True
|
||
)
|
||
subprocess.run(["sudo", "ufw", "allow", "OpenSSH"], check=True)
|
||
self.current_port = self.new_port
|
||
self.new_port = None
|
||
subprocess.run(["sudo", "systemctl", "restart", "sshd"], check=True)
|
||
self.d.msgbox(f"Порт SSH успешно изменен на {self.current_port}.")
|
||
except subprocess.CalledProcessError as e:
|
||
logging.error(f"Ошибка при изменении порта SSH: {e}")
|
||
self.d.msgbox(f"Ошибка при изменении порта SSH: {e}")
|
||
|
||
if self.new_auth_method:
|
||
try:
|
||
auth_config = ""
|
||
if self.new_auth_method == "Только по паролю":
|
||
auth_config = (
|
||
"PasswordAuthentication yes\nPubkeyAuthentication no\n"
|
||
)
|
||
elif self.new_auth_method == "Только по ключу":
|
||
auth_config = (
|
||
"PasswordAuthentication no\nPubkeyAuthentication yes\n"
|
||
)
|
||
elif self.new_auth_method == "По ключу и паролю":
|
||
auth_config = (
|
||
"PasswordAuthentication yes\nPubkeyAuthentication yes\n"
|
||
)
|
||
|
||
os.makedirs(os.path.dirname(SSH_AUTH_CONFIG_PATH), exist_ok=True)
|
||
with open(SSH_AUTH_CONFIG_PATH, "w") as f:
|
||
f.write(auth_config)
|
||
|
||
self.current_auth_method = self.new_auth_method
|
||
self.new_auth_method = None
|
||
subprocess.run(["sudo", "systemctl", "restart", "sshd"], check=True)
|
||
self.d.msgbox("Метод аутентификации SSH успешно изменен.")
|
||
except Exception as e:
|
||
logging.error(f"Ошибка при изменении метода аутентификации: {e}")
|
||
self.d.msgbox(f"Ошибка при изменении метода аутентификации: {e}")
|
||
|
||
def main_menu(self):
|
||
while True:
|
||
status = self.get_sshd_status()
|
||
port_display = self.new_port if self.new_port else self.current_port
|
||
port_status = " (не применено)" if self.new_port else ""
|
||
auth_method_display = (
|
||
self.new_auth_method
|
||
if self.new_auth_method
|
||
else self.current_auth_method
|
||
)
|
||
auth_status = " (не применено)" if self.new_auth_method else ""
|
||
|
||
code, tag = self.d.menu(
|
||
f"Статус sshd: {status}\n"
|
||
f"Текущий порт: {port_display}{port_status}\n"
|
||
f"Метод аутентификации: {auth_method_display}{auth_status}",
|
||
choices=[
|
||
("1", "Включить sshd" if status != "active" else "Выключить sshd"),
|
||
("2", "Изменить порт SSH"),
|
||
("3", "Изменить метод аутентификации"),
|
||
],
|
||
)
|
||
if code != self.d.OK:
|
||
break
|
||
if tag == "1":
|
||
self.toggle_sshd_service("start" if status != "active" else "stop")
|
||
elif tag == "2":
|
||
self.change_ssh_port()
|
||
elif tag == "3":
|
||
self.change_auth_method()
|
||
|
||
return self.new_port is not None or self.new_auth_method is not None
|
||
|
||
|
||
def run_ssh_config():
|
||
ssh_config = SSHConfig()
|
||
changes_made = ssh_config.main_menu()
|
||
return ssh_config, changes_made
|
||
|
||
|
||
if __name__ == "__main__":
|
||
ssh_config, changes_made = run_ssh_config()
|
||
# Note: Changes will be applied in the main menu when "Save and exit" is selected
|