paase
dbb30b70f5
- replace CLI menu with multi-panel Textual interface (sidebar & data table) - implement asynchronous loading for domains and DNS records via worker threads - add support for optional 2FA/TOTP auto-unlock and manual TAN modal - introduce .env configuration fallback for stored credentials - add modals for creating, editing, and deleting DNS records - enable inline row editing via mouse double-click and keyboard navigation - update project documentation in README.md
420 lines
12 KiB
Python
420 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""INWX DNS TUI Client
|
||
Author: Pascal Bouquet <pascal@ptbos.de>
|
||
License: GNU General Public License v3.0 (GPL-3.0)
|
||
Year: 2026
|
||
"""
|
||
|
||
import pyotp
|
||
from textual.app import App, ComposeResult
|
||
from textual.containers import Horizontal, Vertical
|
||
from textual.widgets import Header, Footer, Button, DataTable, Static, OptionList, Label
|
||
from textual.widgets.option_list import Option
|
||
from textual import work
|
||
|
||
import inwx_api as api
|
||
from tui_modals import LoginModal, TwoFactorModal, RecordEditModal
|
||
|
||
class InwxTUIApp(App):
|
||
CSS = """
|
||
Screen {
|
||
layout: vertical;
|
||
background: $surface;
|
||
}
|
||
|
||
#main_container {
|
||
height: 1fr;
|
||
}
|
||
|
||
#sidebar {
|
||
width: 36;
|
||
min-width: 32;
|
||
background: $panel;
|
||
border-right: solid $primary-background;
|
||
}
|
||
|
||
#sidebar_title {
|
||
width: 100%;
|
||
height: 3;
|
||
content-align: center middle;
|
||
text-style: bold;
|
||
background: $boost;
|
||
color: $accent;
|
||
border-bottom: solid $primary-background;
|
||
}
|
||
|
||
#domain_list {
|
||
height: 1fr;
|
||
border: none;
|
||
background: $panel;
|
||
}
|
||
|
||
#domain_list:focus {
|
||
border: none;
|
||
}
|
||
|
||
#content_area {
|
||
width: 1fr;
|
||
height: 1fr;
|
||
}
|
||
|
||
#table_header_info {
|
||
width: 100%;
|
||
height: 3;
|
||
content-align: left middle;
|
||
padding-left: 2;
|
||
background: $boost;
|
||
color: $text;
|
||
text-style: bold;
|
||
border-bottom: solid $primary-background;
|
||
}
|
||
|
||
DataTable {
|
||
height: 1fr;
|
||
border: none;
|
||
}
|
||
|
||
#action_bar {
|
||
height: 3;
|
||
background: $boost;
|
||
align: left middle;
|
||
padding: 0 1;
|
||
border-top: solid $primary-background;
|
||
}
|
||
|
||
.action-btn {
|
||
margin-right: 1;
|
||
height: 1;
|
||
min-width: 16;
|
||
border: none;
|
||
}
|
||
|
||
#status_bar {
|
||
height: 1;
|
||
background: $surface-darken-2;
|
||
color: $text-muted;
|
||
padding-left: 1;
|
||
}
|
||
|
||
.dialog-box {
|
||
padding: 1 2;
|
||
background: $surface;
|
||
border: thick $primary;
|
||
width: 65;
|
||
height: auto;
|
||
align: center middle;
|
||
}
|
||
|
||
.dialog-title {
|
||
text-style: bold;
|
||
color: $accent;
|
||
margin-bottom: 1;
|
||
width: 100%;
|
||
text-align: center;
|
||
}
|
||
|
||
.field-label {
|
||
color: $text-muted;
|
||
margin-top: 1;
|
||
}
|
||
|
||
.type-row {
|
||
height: 3;
|
||
align: left middle;
|
||
margin-bottom: 1;
|
||
}
|
||
|
||
#rec_type_select {
|
||
width: 22;
|
||
margin-right: 1;
|
||
}
|
||
|
||
#rec_type_input {
|
||
width: 1fr;
|
||
margin-bottom: 0;
|
||
}
|
||
|
||
.dialog-buttons {
|
||
height: 3;
|
||
margin-top: 1;
|
||
align: right middle;
|
||
}
|
||
|
||
.modal-btn {
|
||
min-width: 14;
|
||
height: 3;
|
||
margin-left: 1;
|
||
}
|
||
|
||
Input {
|
||
margin-bottom: 0;
|
||
}
|
||
"""
|
||
|
||
BINDINGS = [
|
||
("q", "quit", "Beenden"),
|
||
("r", "refresh_current", "Aktualisieren"),
|
||
("a", "add_record", "Neu"),
|
||
("e", "edit_record", "Bearbeiten"),
|
||
("d", "delete_record", "Löschen"),
|
||
]
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.current_domain = ""
|
||
self.records_cache = {}
|
||
|
||
def compose(self) -> ComposeResult:
|
||
yield Header(show_clock=True)
|
||
|
||
with Horizontal(id="main_container"):
|
||
with Vertical(id="sidebar"):
|
||
yield Label("🌐 Domains", id="sidebar_title")
|
||
yield OptionList(id="domain_list")
|
||
|
||
with Vertical(id="content_area"):
|
||
yield Label("Keine Domain gewählt", id="table_header_info")
|
||
yield DataTable(id="record_table", cursor_type="row")
|
||
|
||
with Horizontal(id="action_bar"):
|
||
yield Button("➕ Hinzufügen (a)", variant="success", id="btn_add", classes="action-btn")
|
||
yield Button("✏️ Bearbeiten (e)", variant="warning", id="btn_edit", classes="action-btn")
|
||
yield Button("🗑️ Löschen (d)", variant="error", id="btn_delete", classes="action-btn")
|
||
|
||
yield Static("Initialisiere...", id="status_bar")
|
||
yield Footer()
|
||
|
||
def on_mount(self) -> None:
|
||
table = self.query_one("#record_table", DataTable)
|
||
table.add_columns("ID", "Name/Host", "Typ", "TTL", "Inhalt/Content")
|
||
|
||
user, pwd, secret = api.get_stored_credentials()
|
||
if user and pwd:
|
||
self.set_status("Melde über .env an...")
|
||
self.async_login(user, pwd, secret)
|
||
else:
|
||
self.prompt_login_modal()
|
||
|
||
def set_status(self, text: str) -> None:
|
||
try:
|
||
self.query_one("#status_bar", Static).update(text)
|
||
except Exception:
|
||
pass
|
||
|
||
@work(thread=True)
|
||
def async_login(self, user, pwd, secret=None):
|
||
status, msg = api.login(user, pwd)
|
||
|
||
if status == "SUCCESS":
|
||
self.app.call_from_thread(self.set_status, "✅ Angemeldet. Lade Domains...")
|
||
self.async_load_domains()
|
||
|
||
elif status == "NEEDS_2FA":
|
||
if secret:
|
||
try:
|
||
totp = pyotp.TOTP(secret)
|
||
tan = totp.now()
|
||
ok, unlock_msg = api.unlock_2fa(tan)
|
||
if ok:
|
||
self.app.call_from_thread(self.set_status, "✅ 2FA via Secret bestätigt.")
|
||
self.async_load_domains()
|
||
return
|
||
except Exception:
|
||
pass
|
||
self.app.call_from_thread(self.prompt_2fa_modal)
|
||
|
||
else:
|
||
self.app.call_from_thread(self.set_status, f"❌ Login fehlgeschlagen: {msg}")
|
||
self.app.call_from_thread(self.prompt_login_modal)
|
||
|
||
def prompt_login_modal(self) -> None:
|
||
def on_login_submitted(creds):
|
||
if not creds:
|
||
self.set_status("Abgebrochen.")
|
||
return
|
||
u, p = creds
|
||
_, _, secret = api.get_stored_credentials()
|
||
self.async_login(u, p, secret)
|
||
|
||
self.push_screen(LoginModal(), on_login_submitted)
|
||
|
||
def prompt_2fa_modal(self) -> None:
|
||
def on_2fa_submitted(tan):
|
||
if not tan:
|
||
self.set_status("2FA abgebrochen.")
|
||
return
|
||
self.async_unlock(tan)
|
||
|
||
self.push_screen(TwoFactorModal(), on_2fa_submitted)
|
||
|
||
@work(thread=True)
|
||
def async_unlock(self, tan):
|
||
ok, msg = api.unlock_2fa(tan)
|
||
if ok:
|
||
self.app.call_from_thread(self.set_status, "✅ 2FA entsperrt.")
|
||
self.async_load_domains()
|
||
else:
|
||
self.app.call_from_thread(self.set_status, f"❌ 2FA-Fehler: {msg}")
|
||
self.app.call_from_thread(self.prompt_2fa_modal)
|
||
|
||
@work(thread=True)
|
||
def async_load_domains(self):
|
||
ok, domains = api.get_domain_list()
|
||
|
||
def update_ui():
|
||
try:
|
||
d_list = self.query_one("#domain_list", OptionList)
|
||
d_list.clear_options()
|
||
if ok and domains:
|
||
for d in domains:
|
||
d_list.add_option(Option(prompt=d, id=d))
|
||
d_list.highlighted = 0
|
||
self.set_status(f"✅ {len(domains)} Domains geladen.")
|
||
self.async_load_dns_records(domains[0])
|
||
elif ok:
|
||
self.set_status("Keine Domains vorhanden.")
|
||
else:
|
||
self.set_status(f"❌ Fehler: {domains}")
|
||
except Exception as e:
|
||
self.set_status(f"UI-Fehler: {e}")
|
||
|
||
self.app.call_from_thread(update_ui)
|
||
|
||
def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None:
|
||
selected_domain = str(event.option_id)
|
||
self.async_load_dns_records(selected_domain)
|
||
|
||
@work(thread=True)
|
||
def async_load_dns_records(self, domain: str):
|
||
self.current_domain = domain
|
||
|
||
def set_loading():
|
||
try:
|
||
self.query_one("#table_header_info", Label).update(f"DNS-Zone: [b]{domain}[/b]")
|
||
self.set_status(f"Lade Einträge für {domain}...")
|
||
except Exception:
|
||
pass
|
||
|
||
self.app.call_from_thread(set_loading)
|
||
|
||
ok, data = api.get_dns_records(domain)
|
||
|
||
def update_table():
|
||
try:
|
||
table = self.query_one("#record_table", DataTable)
|
||
table.clear()
|
||
self.records_cache.clear()
|
||
|
||
if ok:
|
||
for rec in data:
|
||
r_id = str(rec.get("id"))
|
||
self.records_cache[r_id] = rec
|
||
table.add_row(
|
||
r_id,
|
||
rec.get("name", "@"),
|
||
rec.get("type", ""),
|
||
str(rec.get("ttl", "")),
|
||
rec.get("content", ""),
|
||
key=r_id
|
||
)
|
||
self.set_status(f"✅ {len(data)} Records für {domain} geladen.")
|
||
else:
|
||
self.set_status(f"❌ Fehler: {data}")
|
||
except Exception as e:
|
||
self.set_status(f"Tabelle konnte nicht aktualisiert werden: {e}")
|
||
|
||
self.app.call_from_thread(update_table)
|
||
|
||
def get_selected_record(self) -> dict | None:
|
||
table = self.query_one("#record_table", DataTable)
|
||
if table.row_count == 0 or table.cursor_coordinate is None:
|
||
return None
|
||
cell_key = table.coordinate_to_cell_key(table.cursor_coordinate)
|
||
return self.records_cache.get(cell_key.row_key.value)
|
||
|
||
def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
|
||
self.action_edit_record()
|
||
|
||
def on_button_pressed(self, event: Button.Pressed) -> None:
|
||
btn_id = event.button.id
|
||
if btn_id == "btn_add":
|
||
self.action_add_record()
|
||
elif btn_id == "btn_edit":
|
||
self.action_edit_record()
|
||
elif btn_id == "btn_delete":
|
||
self.action_delete_record()
|
||
|
||
def action_refresh_current(self) -> None:
|
||
if self.current_domain:
|
||
self.async_load_dns_records(self.current_domain)
|
||
else:
|
||
self.async_load_domains()
|
||
|
||
# --- Asynchrone Worker-Methoden ---
|
||
|
||
@work(thread=True)
|
||
def async_add_record(self, domain: str, data: dict) -> None:
|
||
ok, msg = api.add_dns_record(
|
||
domain, data["name"], data["type"], data["content"], data["ttl"]
|
||
)
|
||
self.app.call_from_thread(
|
||
self.set_status, "✅ Record hinzugefügt." if ok else f"❌ Fehler: {msg}"
|
||
)
|
||
if ok:
|
||
self.async_load_dns_records(domain)
|
||
|
||
@work(thread=True)
|
||
def async_edit_record(self, record_id: str, data: dict) -> None:
|
||
ok, msg = api.update_dns_record(record_id, data["content"], data["ttl"])
|
||
self.app.call_from_thread(
|
||
self.set_status, f"✅ Record {record_id} aktualisiert." if ok else f"❌ Fehler: {msg}"
|
||
)
|
||
if ok:
|
||
self.async_load_dns_records(self.current_domain)
|
||
|
||
@work(thread=True)
|
||
def async_delete_record(self, record_id: str) -> None:
|
||
ok, msg = api.delete_dns_record(record_id)
|
||
self.app.call_from_thread(
|
||
self.set_status, f"✅ Record {record_id} gelöscht." if ok else f"❌ Fehler: {msg}"
|
||
)
|
||
if ok:
|
||
self.async_load_dns_records(self.current_domain)
|
||
|
||
# --- UI Aktionen ---
|
||
|
||
def action_add_record(self) -> None:
|
||
if not self.current_domain:
|
||
self.set_status("Erst Domain links auswählen.")
|
||
return
|
||
|
||
def handle_add(data):
|
||
if data:
|
||
self.async_add_record(self.current_domain, data)
|
||
|
||
self.push_screen(RecordEditModal(default_domain=self.current_domain), handle_add)
|
||
|
||
def action_edit_record(self) -> None:
|
||
rec = self.get_selected_record()
|
||
if not rec:
|
||
self.set_status("Kein Record ausgewählt.")
|
||
return
|
||
|
||
def handle_edit(data):
|
||
if data:
|
||
self.async_edit_record(rec["id"], data)
|
||
|
||
self.push_screen(RecordEditModal(record=rec), handle_edit)
|
||
|
||
def action_delete_record(self) -> None:
|
||
rec = self.get_selected_record()
|
||
if not rec:
|
||
self.set_status("Kein Record ausgewählt.")
|
||
return
|
||
|
||
self.async_delete_record(rec["id"])
|
||
|
||
def action_quit(self) -> None:
|
||
api.logout()
|
||
self.exit()
|