507 lines
No EOL
20 KiB
Python
507 lines
No EOL
20 KiB
Python
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() |