diff --git a/.env b/.env new file mode 100644 index 0000000..51c1000 --- /dev/null +++ b/.env @@ -0,0 +1,8 @@ +INWX_USER=dein_nutzername +INWX_PASS=dein_passwort + +# Optional: Trage hier dein 2FA/TOTP-Secret ein (Base32), +# damit der 2FA-Code automatisch erzeugt wird und keine Abfrage nötig ist. +# Wenn leer, fragt das TUI dich nach dem Code aus deiner Authenticator-App. +INWX_2FA_SECRET= + diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/README.md b/README.md index e6ed673..68f6530 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,51 @@ -# inwx_tui +# INWX DNS TUI Client + +Python ane **Textual** no upyog kari banavavama aavelo ek aadhunik, clickable Terminal User Interface (TUI) client, jena thi tame INWX na DNS records ane domains ne sahelta thi manage kari shako chho. + +## Features + +- **Clickable GUI in Terminal**: Mouse ane keyboard shortcuts banne dwara chalavi shakay chhe. +- **Domain Overview**: Login thaya pachhi account ma rahela badha domains automatic left sidebar ma load thai jaay chhe. +- **DNS Records Management**: + - DNS records jova (A, AAAA, CNAME, MX, TXT, SRV, CAA, NS, PTR, vagere). + - Double-click athva `Enter` dabavi ne record edit karvo. + - Nava records dropdown menu ane custom input sathe add karva. + - Records delete ane live refresh karva. +- **2FA / TOTP Support**: + - Jo account ma 2FA chalu hoy to j code mangshe. + - `.env` ma TOTP secret mukine automatic unlock pan kari shakay chhe. +- **Non-blocking Workers**: Background async workers lidhe API call vakhte UI freeze thatu nathi. + +--- + +## File Structure + +```text +inwx_tui/ +├── .env # INWX API credentials ane config +├── inwx_api.py # INWX JSON-RPC API logic ane session management +├── tui_modals.py # Modal popups (Login, 2FA prompt, Record Edit/Add) +├── tui_app.py # Textual TUI main interface ane event handling +├── main.py # Application start karvani entrypoint script +└── README.md # Project documentation + +## Requirements + +Python 3.10+ + +Linux / macOS / WSL (Windows Terminal) + +´´´bash +pip install textual python-dotenv requests pyotp + +## Usage + +```bash +python3 main.py +``` +Or, by granting execution permission: + +```bash +chmod +x main.py +./main.py diff --git a/__pycache__/inwx_api.cpython-313.pyc b/__pycache__/inwx_api.cpython-313.pyc new file mode 100644 index 0000000..1e271a5 Binary files /dev/null and b/__pycache__/inwx_api.cpython-313.pyc differ diff --git a/__pycache__/tui_app.cpython-313.pyc b/__pycache__/tui_app.cpython-313.pyc new file mode 100644 index 0000000..4ea933e Binary files /dev/null and b/__pycache__/tui_app.cpython-313.pyc differ diff --git a/__pycache__/tui_modals.cpython-313.pyc b/__pycache__/tui_modals.cpython-313.pyc new file mode 100644 index 0000000..59277e8 Binary files /dev/null and b/__pycache__/tui_modals.cpython-313.pyc differ diff --git a/inwx_api.py b/inwx_api.py new file mode 100644 index 0000000..586b987 --- /dev/null +++ b/inwx_api.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +"""INWX DNS TUI Client +Author: Pascal Bouquet +License: GNU General Public License v3.0 (GPL-3.0) +Year: 2026 +""" + +import os +import json +import requests +from dotenv import load_dotenv + +load_dotenv() + +API_ENDPOINT = "https://api.domrobot.com/jsonrpc/" +SESSION_ID = None + +def get_stored_credentials(): + user = os.getenv("INWX_USER", "").strip() + pwd = os.getenv("INWX_PASS", "").strip() + secret = os.getenv("INWX_2FA_SECRET", "").strip() + + placeholders = {"", "username", "dein_nutzername", "dein_benutzername", "example_user", "password", "passwort"} + if user.lower() in placeholders or pwd.lower() in placeholders: + return None, None, secret + return user, pwd, secret + +def api_call(method, params=None): + global SESSION_ID + if params is None: + params = {} + + headers = {"Content-Type": "application/json"} + if SESSION_ID: + headers["Cookie"] = f"domrobot={SESSION_ID}" + + payload = { + "jsonrpc": "2.0", + "method": method, + "params": params, + "id": 1 + } + + try: + response = requests.post(API_ENDPOINT, headers=headers, data=json.dumps(payload), timeout=12) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + return {"code": -1, "msg": f"Netzwerkfehler: {e}"} + +def login(user, password): + global SESSION_ID + headers = {"Content-Type": "application/json"} + payload = { + "jsonrpc": "2.0", + "method": "account.login", + "params": {"user": user, "pass": password}, + "id": 1 + } + + try: + response = requests.post(API_ENDPOINT, headers=headers, data=json.dumps(payload), timeout=12) + response.raise_for_status() + result = response.json() + + if result.get("code") == 1000: + if "domrobot" in response.cookies: + SESSION_ID = response.cookies["domrobot"] + else: + return "ERROR", "Kein Session-Cookie von INWX erhalten." + + res_data = result.get("resData", {}) + tfa_val = res_data.get("tfa") + + # INWX liefert bei inaktivem 2FA None, 0, "0" oder "NONE" + if tfa_val in (0, "0", None, False, "", "NONE", "none"): + return "SUCCESS", "Login erfolgreich." + + return "NEEDS_2FA", f"2FA erforderlich ({tfa_val})." + + return "ERROR", result.get("msg", "Zugangsdaten ungültig.") + except requests.exceptions.RequestException as e: + return "ERROR", str(e) + +def unlock_2fa(tan_code): + res = api_call("account.unlock", {"tan": str(tan_code).strip()}) + if res.get("code") == 1000: + return True, "2FA erfolgreich entsperrt." + return False, res.get("msg", "2FA-Code ungültig.") + +def logout(): + global SESSION_ID + if SESSION_ID: + api_call("account.logout") + SESSION_ID = None + +def get_domain_list(): + res = api_call("nameserver.list") + if res.get("code") == 1000: + domains_data = res.get("resData", {}).get("domains", []) + return True, [d.get("domain") for d in domains_data if "domain" in d] + return False, res.get("msg", "Fehler beim Abrufen der Domainliste.") + +def get_dns_records(domain): + res = api_call("nameserver.info", {"domain": domain}) + if res.get("code") == 1000: + return True, res.get("resData", {}).get("record", []) + return False, res.get("msg", "Fehler beim Laden der Records.") + +def add_dns_record(domain, name, record_type, content, ttl): + params = { + "domain": domain, + "name": name, + "type": record_type, + "content": content, + "ttl": int(ttl) + } + res = api_call("nameserver.createRecord", params) + return res.get("code") == 1000, res.get("msg", "OK") + +def update_dns_record(record_id, content=None, ttl=None): + params = {"id": int(record_id)} + if content: + params["content"] = content + if ttl: + params["ttl"] = int(ttl) + res = api_call("nameserver.updateRecord", params) + return res.get("code") == 1000, res.get("msg", "OK") + +def delete_dns_record(record_id): + res = api_call("nameserver.deleteRecord", {"id": int(record_id)}) + return res.get("code") == 1000, res.get("msg", "OK") diff --git a/main.py b/main.py new file mode 100755 index 0000000..26e3acf --- /dev/null +++ b/main.py @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 +"""INWX DNS TUI Client +Author: Pascal Bouquet +License: GNU General Public License v3.0 (GPL-3.0) +Year: 2026 +""" + +import sys +from tui_app import InwxTUIApp + +def main(): + app = InwxTUIApp() + try: + app.run() + except KeyboardInterrupt: + sys.exit(0) + +if __name__ == "__main__": + main() diff --git a/tui_app.py b/tui_app.py new file mode 100644 index 0000000..3c15913 --- /dev/null +++ b/tui_app.py @@ -0,0 +1,419 @@ +#!/usr/bin/env python3 +"""INWX DNS TUI Client +Author: Pascal Bouquet +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() diff --git a/tui_modals.py b/tui_modals.py new file mode 100644 index 0000000..77fc6f1 --- /dev/null +++ b/tui_modals.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +"""INWX DNS TUI Client +Author: Pascal Bouquet +License: GNU General Public License v3.0 (GPL-3.0) +Year: 2026 +""" + +from textual.app import ComposeResult +from textual.containers import Vertical, Horizontal +from textual.widgets import Label, Input, Button, Select +from textual.screen import ModalScreen + +RECORD_TYPES = [ + ("A", "A"), + ("AAAA", "AAAA"), + ("CNAME", "CNAME"), + ("MX", "MX"), + ("TXT", "TXT"), + ("SRV", "SRV"), + ("CAA", "CAA"), + ("NS", "NS"), + ("PTR", "PTR"), + ("TLSA", "TLSA"), +] + +class LoginModal(ModalScreen[tuple[str, str] | None]): + """Login-Maske bei fehlender oder fehlerhafter .env.""" + def compose(self) -> ComposeResult: + with Vertical(classes="dialog-box"): + yield Label("🔐 INWX Login", classes="dialog-title") + yield Input(placeholder="Benutzername", id="user_input") + yield Input(placeholder="Passwort", password=True, id="pass_input") + with Horizontal(classes="dialog-buttons"): + yield Button("Einloggen", variant="primary", id="btn_submit", classes="modal-btn") + yield Button("Abbrechen", variant="error", id="btn_cancel", classes="modal-btn") + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "btn_submit": + user = self.query_one("#user_input", Input).value.strip() + pwd = self.query_one("#pass_input", Input).value.strip() + self.dismiss((user, pwd)) + else: + self.dismiss(None) + + +class TwoFactorModal(ModalScreen[str | None]): + """Fragt den 6-stelligen 2FA / TOTP-Code ab.""" + def compose(self) -> ComposeResult: + with Vertical(classes="dialog-box"): + yield Label("🛡️ 2FA-Bestätigung", classes="dialog-title") + yield Label("6-stelligen TAN-Code eingeben:") + yield Input(placeholder="z.B. 123456", id="tan_input", max_length=6) + with Horizontal(classes="dialog-buttons"): + yield Button("Bestätigen", variant="primary", id="btn_submit_tan", classes="modal-btn") + yield Button("Abbrechen", variant="error", id="btn_cancel", classes="modal-btn") + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "btn_submit_tan": + tan = self.query_one("#tan_input", Input).value.strip() + self.dismiss(tan) + else: + self.dismiss(None) + + +class RecordEditModal(ModalScreen[dict | None]): + """Dialog zum Anlegen und Bearbeiten von DNS-Records.""" + def __init__(self, record: dict | None = None, default_domain: str = ""): + super().__init__() + self.record = record + self.default_domain = default_domain + + def compose(self) -> ComposeResult: + is_edit = self.record is not None + title = f"Record ID {self.record['id']} bearbeiten" if is_edit else f"Neuer Record für {self.default_domain}" + + with Vertical(classes="dialog-box"): + yield Label(title, classes="dialog-title") + + if not is_edit: + yield Label("Name / Host (@ oder z.B. mail):", classes="field-label") + yield Input(placeholder="@", value="@", id="rec_name") + + yield Label("Record-Typ (Auswählen oder eintippen):", classes="field-label") + with Horizontal(classes="type-row"): + yield Select( + RECORD_TYPES, + value="A", + prompt="Typ wählen", + id="rec_type_select", + allow_blank=False + ) + yield Input(value="A", placeholder="Typ", id="rec_type_input") + + yield Label("Inhalt / Ziel:", classes="field-label") + val_content = str(self.record.get("content", "")) if is_edit else "" + yield Input(placeholder="z.B. IP, mx01.domain.tld, Text", value=val_content, id="rec_content") + + yield Label("TTL (Sekunden):", classes="field-label") + val_ttl = str(self.record.get("ttl", "3600")) if is_edit else "3600" + yield Input(placeholder="3600", value=val_ttl, id="rec_ttl") + + with Horizontal(classes="dialog-buttons"): + yield Button("Speichern", variant="success", id="btn_save", classes="modal-btn") + yield Button("Abbrechen", variant="error", id="btn_cancel", classes="modal-btn") + + def on_select_changed(self, event: Select.Changed) -> None: + """Überträgt die Dropdown-Auswahl direkt in das Text-Eingabefeld.""" + if event.select.id == "rec_type_select" and event.value != Select.BLANK: + type_input = self.query_one("#rec_type_input", Input) + type_input.value = str(event.value) + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "btn_save": + content = self.query_one("#rec_content", Input).value.strip() + ttl = self.query_one("#rec_ttl", Input).value.strip() or "3600" + + if self.record: + data = {"content": content, "ttl": ttl} + else: + name = self.query_one("#rec_name", Input).value.strip() + rec_type = self.query_one("#rec_type_input", Input).value.strip().upper() + data = {"name": name, "type": rec_type, "content": content, "ttl": ttl} + self.dismiss(data) + else: + self.dismiss(None)