import subprocess from textual import log from textual import on from textual.app import App, ComposeResult from textual.containers import Container from textual.screen import Screen from textual.widgets import Header, Footer, Button, Static, Input, RadioButton, RadioSet import setuptools import importlib.resources class CustomRadioButton(RadioButton): BUTTON_INNER = "x" class DHCPScreen(Screen): def compose(self) -> ComposeResult: yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Введите настройки сети:", classes="chooseBlock__title") yield Input(placeholder="Введите DNS", id="input_dns") yield Button("Протестировать", id="button__testDHCP", classes="chooseBlock__button") yield Button("Сохранить и выйти", id="button__saveDHCP", classes="chooseBlock__button") yield Footer() class StaticScreen(Screen): def compose(self) -> ComposeResult: yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Введите настройки сети:", classes="chooseBlock__title") yield Input(placeholder="Введите IP адрес", id="input_ip") yield Input(placeholder="Введите маску", id="input_mask") yield Input(placeholder="Введите шлюз", id="input_gateway") yield Button("Протестировать", id="button__testStatic", classes="chooseBlock__button") yield Button("Сохранить и выйти", id="button__saveStatic", classes="chooseBlock__button") yield Footer() class ChooseTypeScreen(Screen): def compose(self) -> ComposeResult: yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Выберите тип подключения к сети", classes="chooseBlock__title") yield Static(f"Интерфейс: {self.app.get_interface()}", classes="chooseBlock__title") yield Button("DHCP", id="button__dhcp", classes="chooseBlock__button") yield Button("Static", id="button__static", classes="chooseBlock__button") yield Footer() class ChooseInterfaceScreen(Screen): def compose(self) -> ComposeResult: self.interfaces = self.get_network_interfaces() yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Выберите интерфейс", classes="chooseBlock__title") with RadioSet(id="radio__interface"): if len(self.interfaces) != 0: for interface in self.interfaces: yield CustomRadioButton(interface) yield Button("Выбрать", id="button__chooseInterface", classes="chooseBlock__button") yield Button("На главную", id="button__toOptions", classes="chooseBlock__button") yield Footer() @staticmethod def get_network_interfaces(): interfaces = str.split(subprocess.run(["ls", "/sys/class/net/"], capture_output=True, text=True).stdout, "\n") return interfaces[:-1] class ChooseOptionScreen(Screen): def update_options(self, new_options): log(12) self.options = new_options self.refresh() def compose(self) -> ComposeResult: log(123123) self.options = self.app.get_options() yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Выберите действие", classes="chooseBlock__title") with RadioSet(id="radio__option"): for option in range(len(self.options)): yield CustomRadioButton( f"{option + 1}) {self.options[option]['name']}{self.options[option]['tabs']}{self.options[option]['value']}", id=f"radio__option_{option}") yield Button("Выбрать", id="button__chooseOption", classes="chooseBlock__button") yield Footer() class DomainScreen(Screen): def compose(self) -> ComposeResult: yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Введите новый домен:", classes="chooseBlock__title") yield Input(placeholder="Введите домен", id="input_domain") yield Button("Сохранить и выйти", id="button__domain", classes="chooseBlock__button") yield Footer() class AddUserScreen(Screen): def compose(self) -> ComposeResult: yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Введите данные для нового пользователя:", classes="chooseBlock__title") yield Input(placeholder="Введите имя пользователя в системе", id="input_user_username") yield Input(placeholder="Введите пароль пользователя", id="input_user_password") yield Input(placeholder="Введите полное имя пользователя", id="input_user_fullname") yield Input(placeholder="Введите мобильный телефон пользователя", id="input_user_mobilePhone") yield Input(placeholder="Введите городской телефон пользователя", id="input_user_cityPhone") yield Input(placeholder="Введите кабинет", id="input_user_cabinet") yield Button("Создать", id="button__addUser", classes="chooseBlock__button") yield Button("На главную", id="button__toOptions", classes="chooseBlock__button") yield Footer() class SudoUserScreen(Screen): def compose(self) -> ComposeResult: yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Пользователь будет администратором?", classes="chooseBlock__title") yield Button("Да", id="button__sudoYes", classes="chooseBlock__button") yield Button("Нет", id="button__sudoNo", classes="chooseBlock__button") yield Footer() class UpdateScreen(Screen): def compose(self) -> ComposeResult: yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Вы действительно хотите обновить систему?", classes="chooseBlock__title") yield Button("Да", id="button__updateYes", classes="chooseBlock__button") yield Button("Нет", id="button__toOptions", classes="chooseBlock__button") yield Footer() class ChangeSshPortScreen(Screen): def compose(self) -> ComposeResult: yield Header() with Container(classes="chooseBlock centerBlock"): yield Static("Изменение порта ssh.", classes="chooseBlock__title") yield Input(placeholder="Введите новый порт", id="input_ssh") yield Button("Изменить", id="button__ssh", classes="chooseBlock__button") yield Button("На главную", id="button__toOptions", classes="chooseBlock__button") yield Footer() class NetworkApp(App): CSS_PATH = "styles.tcss" BINDINGS = [("Ctrl+C", "action_quit", "Выйти")] SCREENS = { "dhcp": DHCPScreen(), "static": StaticScreen(), "chooseType": ChooseTypeScreen(), "chooseInterface": ChooseInterfaceScreen(), "chooseOption": ChooseOptionScreen(), "chooseDomain": DomainScreen(), "addUser": AddUserScreen(), "updateSystem": UpdateScreen(), "sudoUser": SudoUserScreen(), "changeSshPort": ChangeSshPortScreen(), } def __init__(self): super().__init__() self.interface = None self.selectedOption = None self.userData = { "username": "", "password": "", "fullname": "", "cityPhone": "", "mobilePhone": "", "cabinet": "", } self.sshPort = self.get_sshPort self.options = [ { "name": "Сетевые настройки", "value": "", "tabs": "\t \t", "screen": "chooseInterface" }, { "name": "Доменное имя ПК", "value": self.get_hostname(), "tabs": "\t \t \t", "screen": "chooseDomain" }, { "name": "Добавление пользователя", "value": "", "tabs": "\t \t", "screen": "addUser" }, { "name": "Обновление системы", "value": "", "tabs": "\t \t", "screen": "updateSystem" }, { "name": "Изменить порт ssh", "value": self.sshPort(), "tabs": "\t \t", "screen": "changeSshPort" }, ] @on(Button.Pressed) def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "button__dhcp": self.push_screen("dhcp") elif event.button.id == "button__static": self.push_screen("static") elif event.button.id == "button__saveStatic": self.saveStatic() elif event.button.id == "button__testStatic": self.testStatic() elif event.button.id == "button__saveDHCP": self.saveDHCP() elif event.button.id == "button__testDHCP": self.testDHCP() elif event.button.id == "button__chooseInterface": self.chooseInterface() elif event.button.id == "button__chooseOption": self.chooseOption() elif event.button.id == "button__addUser": self.addUser() elif event.button.id == "button__domain": self.changeDomain() elif event.button.id == "button__updateYes": self.updateSystem() elif event.button.id == "button__toOptions": self.push_screen("chooseOption") elif event.button.id == "button__sudoYes": self.userIsSudo(True) elif event.button.id == "button__sudoNo": self.userIsSudo(False) elif event.button.id == "button__ssh": self.changeSshPort() @on(RadioSet.Changed, "#radio__interface") def on_radioset_changed_interface(self, event: RadioSet.Changed): self.interface = event.pressed.label log(self.interface) @on(RadioSet.Changed, "#radio__option") def on_radioset_changed_option(self, event: RadioSet.Changed): option_index = int(event.pressed.id.split("_")[-1]) self.selectedOption = self.options[option_index] def saveStatic(self) -> None: ip = self.query_one("#input_ip", Input).value mask = self.query_one("#input_mask", Input).value gateway = self.query_one("#input_gateway", Input).value if ip and mask and gateway: interface_config = f""" auto {self.interface} iface {self.interface} inet static address {ip} netmask {mask} gateway {gateway} """ try: with open("/etc/network/interfaces", "r") as f: lines = f.readlines() with open("/etc/network/interfaces", "w") as f: interface_found = False for line in lines: if line.strip() == f"auto {self.interface}": interface_found = True if interface_found and line.strip().startswith("iface"): continue if interface_found and line.strip().startswith( ("address", "netmask", "gateway", "dns-nameservers")): continue if not interface_found or (interface_found and not line.strip()): f.write(line) f.write(interface_config.strip() + "\n") subprocess.run(["sudo", "systemctl", "restart", "networking"], check=True) log(f"Static IP {ip} configured successfully") except subprocess.CalledProcessError as e: log(f"Error configuring static IP: {e}") self.push_screen('chooseInterface') def saveDHCP(self) -> None: dns = self.query_one("#input_dns", Input).value interface_config = f""" auto {self.interface} iface {self.interface} inet dhcp """ try: with open("/etc/network/interfaces", "r") as f: lines = f.readlines() with open("/etc/network/interfaces", "w") as f: interface_found = False for line in lines: if line.strip() == f"auto {self.interface}": interface_found = True if interface_found and line.strip().startswith("iface"): continue if interface_found and line.strip().startswith( ("address", "netmask", "gateway", "dns-nameservers")): continue if not interface_found or (interface_found and not line.strip()): f.write(line) if interface_found and line.strip() == "": interface_found = False f.write(interface_config.strip() + "\n") if dns: dns_entry = f"nameserver {dns}" with open("/etc/resolv.conf", "r") as f: resolv_lines = f.readlines() if dns_entry + "\n" not in resolv_lines: with open("/etc/resolv.conf", "a") as f: f.write(dns_entry + "\n") subprocess.run(["sudo", "systemctl", "restart", "networking"], check=True) log(f"DHCP configured successfully with DNS {dns}") except subprocess.CalledProcessError as e: log(f"Error configuring DHCP: {e}") self.push_screen('chooseInterface') def testStatic(self) -> None: ip = self.query_one("#input_ip", Input).value mask = self.query_one("#input_mask", Input).value gateway = self.query_one("#input_gateway", Input).value if ip and mask and gateway: try: subprocess.run( ["ping", "-c", "3", ip], check=True ) log(f"Ping to {ip} successful") except subprocess.CalledProcessError as e: log(f"Ping to {ip} failed: {e}") def testDHCP(self) -> None: dns = self.query_one("#input_dns", Input).value if dns: try: subprocess.run( ["ping", "-c", "3", dns], check=True ) log(f"Ping to DNS {dns} successful") except subprocess.CalledProcessError as e: log(f"Ping to DNS {dns} failed: {e}") def chooseInterface(self) -> None: if self.interface is not None: self.push_screen('chooseType') def chooseOption(self) -> None: if self.selectedOption is not None: self.push_screen(self.selectedOption['screen']) def changeDomain(self) -> None: domain_input = self.query_one("#input_domain", Input).value if domain_input: try: result = subprocess.run( ["sudo", "hostnamectl", "set-hostname", domain_input], capture_output=True, text=True ) if result.returncode == 0: self.options[1]['value'] = domain_input self.updateOptions() else: log(f"Error changing domain: {result.stderr}") except Exception as e: log(f"Exception changing domain: {e}") def addUser(self) -> None: if (self.query_one("#input_user_username", Input).value != "") and ( self.query_one("#input_user_password", Input).value != ""): for field in self.userData.keys(): self.userData[field] = self.query_one(f"#input_user_{field}", Input).value self.query_one(f"#input_user_{field}", Input).value = "" log(self.userData) self.push_screen("sudoUser") def userIsSudo(self, isAdmin) -> None: try: subprocess.run( ["sudo", "adduser", "--gecos", f"{self.userData['fullname']},{self.userData['cabinet']},{self.userData['cityPhone']},{self.userData['mobilePhone']}", "--disabled-login", self.userData['username']], check=True ) subprocess.run( ["sudo", "chpasswd"], input=f"{self.userData['username']}:{self.userData['password']}", text=True, check=True ) subprocess.run( ["sudo", "mkdir", "-p", f"/home/{self.userData['username']}"], check=True ) subprocess.run( ["sudo", "chown", f"{self.userData['username']}:{self.userData['username']}", f"/home/{self.userData['username']}"], check=True ) subprocess.run( ["sudo", "chmod", "755", f"/home/{self.userData['username']}"], check=True ) subprocess.run( ["sudo", "usermod", "-s", "/bin/bash", self.userData['username']], check=True ) if isAdmin: subprocess.run( ["sudo", "usermod", "-aG", "sudo", self.userData['username']], check=True ) log("User added successfully") except subprocess.CalledProcessError as e: log(f"Error adding user: {e}") self.push_screen("chooseOption") def updateSystem(self) -> None: try: subprocess.run(["sudo", "apt-get", "update"], check=True) subprocess.run(["sudo", "apt-get", "upgrade", "-y"], check=True) log("System updated successfully") except subprocess.CalledProcessError as e: log(f"Error updating system: {e}") def changeSshPort(self) -> None: if self.query_one("#input_ssh", Input).value != "": ssh = self.query_one("#input_ssh", Input).value self.query_one("#input_ssh", Input).value = "" self.options[4]["value"] = ssh try: # First, uncomment the Port line if it is commented and change the port subprocess.run( ["sudo", "sed", "-i", f"/^#Port /c\\Port {ssh}", "/etc/ssh/sshd_config"], check=True ) # Then, ensure that an uncommented Port line is also changed subprocess.run( ["sudo", "sed", "-i", f"/^Port /c\\Port {ssh}", "/etc/ssh/sshd_config"], check=True ) subprocess.run(["sudo", "systemctl", "restart", "ssh"], check=True) log("SSH port changed successfully") except subprocess.CalledProcessError as e: log(f"Error changing SSH port: {e}") self.updateOptions() def get_interface(self): return self.interface def get_options(self): return self.options def on_mount(self) -> None: self.push_screen("chooseOption") def get_hostname(self) -> str: return subprocess.run(["cat", "/etc/hostname"], capture_output=True, text=True).stdout.strip() def get_sshPort(self) -> str: try: result = subprocess.run( ["grep", "^Port", "/etc/ssh/sshd_config"], capture_output=True, text=True ) return result.stdout.strip().split()[-1] except Exception as e: log(f"Exception getting SSH port: {e}") return "22" def updateOptions(self) -> None: self.push_screen("chooseOption") for option in range(len(self.options)): self.query_one( f"#radio__option_{option}").label = f"{option + 1}) {self.options[option]['name']}{self.options[option]['tabs']}{self.options[option]['value']}" def main(): app = NetworkApp() app.run() if __name__ == "__main__": main()