#!/root/sql/vmail-env/bin/python3 import mysql.connector import configparser import socket import sys from argon2 import PasswordHasher, Type from textual.app import App, ComposeResult from textual.widgets import Header, Footer, ListView, ListItem, Label, DataTable, Input, Button, Select from textual.containers import Horizontal, Vertical from textual.screen import ModalScreen # --- Konfiguration & Krypto --- config = configparser.ConfigParser() config.read('config.ini') DEFAULT_DOM = config.get('database', 'default_domain', fallback="") hostname = socket.getfqdn() ph = PasswordHasher(time_cost=3, memory_cost=131072, parallelism=4, hash_len=32, salt_len=16, type=Type.ID) def get_db_connection(): try: return mysql.connector.connect( host=config['database']['host'], user=config['database']['user'], password=config['database']['password'], database=config['database']['database'] ) except Exception as e: print(f"Fehler bei DB-Verbindung: {e}") sys.exit(1) def hash_pw(password): return f"{{ARGON2ID}}{ph.hash(password)}" def fetch_all_domains(): conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT domain FROM domains ORDER BY domain ASC") doms = [r[0] for r in cursor.fetchall()] conn.close() return doms # --- MODAL POPUPS (Eingabemasken & Abfragen) --- class ConfirmationModal(ModalScreen): """Sicherheitsabfrage vor dem Löschen""" def __init__(self, message: str): super().__init__() self.message = message def compose(self) -> ComposeResult: with Vertical(id="dialog"): yield Label(f"[b]⚠️ Sicherheitsabfrage[/b]\n\n{self.message}\n") with Horizontal(id="buttons"): yield Button("Nein", variant="error", id="no") yield Button("Ja", variant="success", id="yes") def on_button_pressed(self, event: Button.Pressed) -> None: self.dismiss(event.button.id == "yes") class EditUserModal(ModalScreen): """Ändert Passwort und Quota zeitgleich""" AUTO_FOCUS = "#quota" def __init__(self, username: str, domain: str, current_quota: str): super().__init__() self.username = username self.domain = domain self.current_quota = current_quota def compose(self) -> ComposeResult: with Vertical(id="dialog"): yield Label(f"[b]Account bearbeiten:[/b] [cyan]{self.username}@{self.domain}[/cyan]\n") yield Label("Quota (in MB):", classes="input-label") yield Input(id="quota", placeholder="Quota", value=self.current_quota) yield Label("Neues Passwort (leer lassen für keine Änderung):", classes="input-label") yield Input(id="password", placeholder="Passwort", password=True) with Horizontal(id="buttons"): yield Button("Abbrechen", variant="error", id="cancel") yield Button("Speichern", variant="success", id="submit") def on_input_submitted(self, event: Input.Submitted) -> None: self.process_submit() def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "submit": self.process_submit() else: self.dismiss(None) def process_submit(self): quota = self.query_one("#quota", Input).value.strip() pw = self.query_one("#password", Input).value self.dismiss({"user": self.username, "dom": self.domain, "pw": pw, "quota": quota}) class EditAliasModal(ModalScreen): """Ändert das Weiterleitungsziel eines Alias""" AUTO_FOCUS = "#dst_user" def __init__(self, alias_id: str, source_email: str, current_dst_user: str, current_dst_dom: str, domains: list): super().__init__() self.alias_id = alias_id self.source_email = source_email self.current_dst_user = current_dst_user self.current_dst_dom = current_dst_dom self.domain_options = [(d, d) for d in domains] def compose(self) -> ComposeResult: with Vertical(id="dialog"): yield Label(f"[b]Alias Ziel ändern für:[/b] [cyan]{self.source_email}[/cyan]\n") yield Label("Ziel Username:", classes="input-label") yield Input(id="dst_user", value=self.current_dst_user) yield Label("Ziel Domain:", classes="input-label") yield Select(self.domain_options, id="dst_dom", value=self.current_dst_dom) with Horizontal(id="buttons"): yield Button("Abbrechen", variant="error", id="cancel") yield Button("Speichern", variant="success", id="submit") def on_input_submitted(self, event: Input.Submitted) -> None: self.process_submit() def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "submit": self.process_submit() else: self.dismiss(None) def process_submit(self): u = self.query_one("#dst_user", Input).value.strip() d = self.query_one("#dst_dom", Select).value if u and d != Select.BLANK: self.dismiss({"id": self.alias_id, "dst_user": u, "dst_dom": d}) else: self.dismiss(None) class EditTLSModal(ModalScreen): """Ändert Policy und Parameter einer TLS-Richtlinie""" def __init__(self, domain: str, current_policy: str, current_params: str): super().__init__() self.domain = domain self.current_policy = current_policy self.current_params = current_params self.p_choices = [("none","none"), ("may","may"), ("encrypt","encrypt"), ("dane","dane"), ("dane-only","dane-only"), ("fingerprint","fingerprint"), ("verify","verify"), ("secure","secure")] def compose(self) -> ComposeResult: with Vertical(id="dialog"): yield Label(f"[b]TLS Policy bearbeiten für:[/b] [cyan]{self.domain}[/cyan]\n") yield Label("Policy:", classes="input-label") yield Select(self.p_choices, id="policy", value=self.current_policy) yield Label("Parameters (optional):", classes="input-label") yield Input(id="params", value=self.current_params) with Horizontal(id="buttons"): yield Button("Abbrechen", variant="error", id="cancel") yield Button("Speichern", variant="success", id="submit") def on_mount(self) -> None: self.query_one("#params", Input).focus() def on_input_submitted(self, event: Input.Submitted) -> None: self.process_submit() def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "submit": self.process_submit() else: self.dismiss(None) def process_submit(self): p = self.query_one("#policy", Select).value param = self.query_one("#params", Input).value.strip() if p != Select.BLANK: self.dismiss({"dom": self.domain, "policy": p, "params": param}) else: self.dismiss(None) class AddUserModal(ModalScreen): AUTO_FOCUS = "#username" def __init__(self, domains): super().__init__() self.domain_options = [(d, d) for d in domains] def compose(self) -> ComposeResult: with Vertical(id="dialog"): yield Label("[b]Neuen E-Mail Account anlegen[/b]\n") yield Input(id="username", placeholder="Username (vor dem @)") yield Select(self.domain_options, id="domain_select", prompt="Domain auswählen...", value=DEFAULT_DOM if DEFAULT_DOM in [d[0] for d in self.domain_options] else Select.BLANK) yield Input(id="password", placeholder="Passwort", password=True) with Horizontal(id="buttons"): yield Button("Abbrechen", variant="error", id="cancel") yield Button("Anlegen", variant="success", id="submit") def on_input_submitted(self, event: Input.Submitted) -> None: self.process_submit() def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "submit": self.process_submit() else: self.dismiss(None) def process_submit(self): user = self.query_one("#username", Input).value.strip() dom = self.query_one("#domain_select", Select).value pw = self.query_one("#password", Input).value if user and dom != Select.BLANK and pw: self.dismiss({"user": user, "dom": dom, "pw": pw}) else: self.dismiss(None) class GenericInputModal(ModalScreen): def __init__(self, title: str, fields: list): super().__init__() self.title_text = title self.fields = fields def compose(self) -> ComposeResult: with Vertical(id="dialog"): yield Label(f"[b]{self.title_text}[/b]\n") for fid, placeholder, is_pw, choices in self.fields: if choices: yield Select(choices, id=fid, prompt=placeholder) else: yield Input(id=fid, placeholder=placeholder, password=is_pw) with Horizontal(id="buttons"): yield Button("Abbrechen", variant="error", id="cancel") yield Button("Bestätigen", variant="success", id="submit") def on_mount(self) -> None: first = self.query("Input, Select").first() if first: first.focus() def on_input_submitted(self, event: Input.Submitted) -> None: self.process_submit() def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "submit": self.process_submit() else: self.dismiss(None) def process_submit(self): data = {fid: self.query_one(f"#{fid}", Select if choices else Input).value for fid, _, _, choices in self.fields} self.dismiss(data) # --- MAIN APP --- class VMailApp(App): CSS = """ Screen { background: $surface; } #sidebar { width: 35; background: $panel; border-right: solid $primary; } #main-content { padding: 1 2; } #status-text { margin-bottom: 1; } ListItem { padding: 0 1; } .input-label { text-align: left; margin: 0 0 0 1; } /* Modals */ ConfirmationModal, AddUserModal, EditUserModal, EditAliasModal, EditTLSModal, GenericInputModal { align: center middle; } #dialog { padding: 1 2; background: $panel; text-align: center; width: 55; border: thick $primary; } #buttons { margin-top: 1; align: center middle; } Button { margin: 0 1; } Input { margin-bottom: 1; } Select { margin-bottom: 1; } """ TITLE = f"VMail Manager ({hostname})" BINDINGS = [ ("q", "quit", "Beenden"), ("escape", "back_to_menu", "Zurück zum Menü") ] def compose(self) -> ComposeResult: yield Header(show_clock=True) with Horizontal(): with Vertical(id="sidebar"): yield Label("[b] NAVIGATION [/b]\n", expand=True) with ListView(id="menu-list"): # Kategorie 1 yield ListItem(Label("[cyan]─ User Verwaltung ─[/cyan]"), disabled=True) yield ListItem(Label(" Anzeigen / ändern"), id="user_show") yield ListItem(Label(" Hinzufügen"), id="user_add") yield ListItem(Label(" Status umschalten"), id="user_toggle") # Kategorie 2 yield ListItem(Label("[cyan]─ Alias Verwaltung ─[/cyan]"), disabled=True) yield ListItem(Label(" Anzeigen / ändern"), id="alias_show") yield ListItem(Label(" Hinzufügen"), id="alias_add") # Kategorie 3 yield ListItem(Label("[cyan]─ Domain Verwaltung ─[/cyan]"), disabled=True) yield ListItem(Label(" Anzeigen"), id="domain_show") yield ListItem(Label(" Hinzufügen"), id="domain_add") # Kategorie 4 yield ListItem(Label("[cyan]─ TLS Verwaltung ─[/cyan]"), disabled=True) yield ListItem(Label(" Anzeigen / ändern"), id="tls_show") yield ListItem(Label(" Hinzufügen"), id="tls_add") with Vertical(id="main-content"): yield Label("Willkommen. Wähle eine Option links mit den Pfeiltasten und Enter.", id="status-text") yield DataTable(id="data-table") yield Footer() def on_mount(self) -> None: self.table = self.query_one("#data-table", DataTable) self.table.cursor_type = "row" self.table.display = False self.current_action = "none" self.query_one("#menu-list", ListView).focus() def action_back_to_menu(self) -> None: self.table.display = False self.query_one("#menu-list", ListView).focus() self.query_one("#status-text", Label).update("Wähle eine Option links.") def refresh_view(self) -> None: self.table.clear(columns=True) status_lbl = self.query_one("#status-text", Label) conn = get_db_connection() cursor = conn.cursor() if self.current_action == "user_show": status_lbl.update("[b]Accounts ([Enter] Bearbeiten | [Entf/Del] Löschen | [Esc] Menü):[/b]") self.table.display = True self.table.add_columns("ID", "Email Adresse", "Status", "Quota") cursor.execute("SELECT id, username, domain, enabled, quota FROM accounts ORDER BY id ASC") for r in cursor.fetchall(): self.table.add_row(str(r[0]), f"{r[1]}@{r[2]}", "AKTIV" if r[3] else "GESPERRT", f"{r[4]} MB") elif self.current_action == "alias_show": status_lbl.update("[b]Aliase ([Enter] Ziel ändern | [Entf/Del] Löschen | [Esc] Menü):[/b]") self.table.display = True self.table.add_columns("ID", "Alias (Source)", "Ziel (Destination)") cursor.execute("SELECT id, source_username, source_domain, destination_username, destination_domain FROM aliases ORDER BY id ASC") for r in cursor.fetchall(): self.table.add_row(str(r[0]), f"{r[1]}@{r[2]}", f"{r[3]}@{r[4]}") elif self.current_action == "domain_show": status_lbl.update("[b]Domains ([Entf/Del] Löschen | [Esc] Menü):[/b]") self.table.display = True self.table.add_columns("ID", "Domain") cursor.execute("SELECT id, domain FROM domains ORDER BY id ASC") for r in cursor.fetchall(): self.table.add_row(str(r[0]), r[1]) elif self.current_action == "tls_show": status_lbl.update("[b]TLS Policies ([Enter] Bearbeiten | [Entf/Del] Löschen | [Esc] Menü):[/b]") self.table.display = True self.table.add_columns("ID", "Domain", "Policy", "Parameters") cursor.execute("SELECT id, domain, policy, params FROM tlspolicies ORDER BY id ASC") for r in cursor.fetchall(): self.table.add_row(str(r[0]), r[1], r[2], r[3] or "-") conn.close() if self.table.display: self.table.focus() def on_list_view_selected(self, event: ListView.Selected) -> None: self.current_action = event.item.id domains = fetch_all_domains() if "show" in self.current_action: self.refresh_view() elif self.current_action == "user_add": self.push_screen(AddUserModal(domains), self.callback_user_add) elif self.current_action == "user_toggle": self.push_screen(GenericInputModal("Status umschalten", [("email", "Ganze E-Mail Adresse", False, None)]), self.callback_user_toggle) elif self.current_action == "alias_add": dom_choices = [(d, d) for d in domains] fields = [("s_u", "Alias Name (vor @)", False, None), ("s_d", "Alias Domain", False, dom_choices), ("d_u", "Ziel Name (vor @)", False, None), ("d_d", "Ziel Domain", False, dom_choices)] self.push_screen(GenericInputModal("Alias hinzufügen", fields), self.callback_alias_add) elif self.current_action == "domain_add": self.push_screen(GenericInputModal("Domain registrieren", [("dom", "Domain Name (z.B. test.de)", False, None)]), self.callback_domain_add) elif self.current_action == "tls_add": p_choices = [("none","none"), ("may","may"), ("encrypt","encrypt"), ("dane","dane"), ("dane-only","dane-only"), ("fingerprint","fingerprint"), ("verify","verify"), ("secure","secure")] fields = [("dom", "Domain (z.B. gmx.de)", False, None), ("pol", "Policy wählen", False, p_choices), ("param", "Params (optional)", False, None)] self.push_screen(GenericInputModal("TLS Policy hinzufügen", fields), self.callback_tls_add) # --- ABFANGEN DER ENTF/DEL TASTE FÜR LÖSCHUNGEN --- def on_key(self, event) -> None: if event.key == "delete" and self.table.has_focus and self.table.display: try: row = self.table.get_row_at(self.table.cursor_row) self.row_to_delete = row if self.current_action == "user_show": self.push_screen(ConfirmationModal(f"Soll der Account '{row[1]}' wirklich\nunwiderruflich gelöscht werden?"), self.callback_del_user_confirmed) elif self.current_action == "alias_show": self.push_screen(ConfirmationModal(f"Soll der Alias ID {row[0]}\n({row[1]} -> {row[2]})\ngelöscht werden?"), self.callback_del_alias_confirmed) elif self.current_action == "domain_show": self.push_screen(ConfirmationModal(f"Soll die Domain '{row[1]}' gelöscht\nwerden?\n(Schlägt fehl, wenn noch User verknüpft sind)"), self.callback_del_domain_confirmed) elif self.current_action == "tls_show": self.push_screen(ConfirmationModal(f"Soll die TLS-Richtlinie für '{row[1]}'\ngelöscht werden?"), self.callback_del_tls_confirmed) except Exception: pass # --- ENTER-TASTE AUF ZEILE (Bearbeiten) --- def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: row = self.table.get_row(event.row_key) if self.current_action == "user_show": email = row[1] quota = row[3].replace(" MB", "") if "@" in email: u, d = email.split("@") self.push_screen(EditUserModal(u, d, quota), self.callback_user_edit_save) elif self.current_action == "alias_show": alias_id = row[0] src = row[1] dst = row[2] if "@" in dst: du, dd = dst.split("@") self.push_screen(EditAliasModal(alias_id, src, du, dd, fetch_all_domains()), self.callback_alias_edit_save) elif self.current_action == "tls_show": dom = row[1] pol = row[2] param = row[3] if row[3] != "-" else "" self.push_screen(EditTLSModal(dom, pol, param), self.callback_tls_edit_save) # --- DATENBANK WRITE CALLBACKS --- def callback_user_add(self, data) -> None: if not data: return conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("INSERT INTO accounts (username, domain, password, quota, enabled) VALUES (%s, %s, %s, 1024, 1)", (data["user"], data["dom"], hash_pw(data["pw"]))) conn.commit() self.query_one("#status-text", Label).update(f"✔ User {data['user']}@{data['dom']} angelegt!") except Exception as e: self.query_one("#status-text", Label).update(f"❌ Fehler: {e}") conn.close() self.query_one("#menu-list", ListView).focus() def callback_user_edit_save(self, data) -> None: if not data: return conn = get_db_connection() cursor = conn.cursor() try: quota = int(data["quota"]) if data["quota"].isdigit() else 1024 cursor.execute("UPDATE accounts SET quota = %s WHERE username = %s AND domain = %s", (quota, data["user"], data["dom"])) if data["pw"]: cursor.execute("UPDATE accounts SET password = %s WHERE username = %s AND domain = %s", (hash_pw(data["pw"]), data["user"], data["dom"])) conn.commit() self.query_one("#status-text", Label).update(f"✔ Account {data['user']}@{data['dom']} erfolgreich aktualisiert.") except Exception as e: self.query_one("#status-text", Label).update(f"❌ Fehler: {e}") conn.close() self.refresh_view() def callback_alias_edit_save(self, data) -> None: if not data: return conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("UPDATE aliases SET destination_username = %s, destination_domain = %s WHERE id = %s", (data["dst_user"], data["dst_dom"], data["id"])) conn.commit() self.query_one("#status-text", Label).update(f"✔ Alias ID {data['id']} aktualisiert.") except Exception as e: self.query_one("#status-text", Label).update(f"❌ Fehler: {e}") conn.close() self.refresh_view() def callback_tls_edit_save(self, data) -> None: if not data: return conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("UPDATE tlspolicies SET policy = %s, params = %s WHERE domain = %s", (data["policy"], data["params"] or None, data["dom"])) conn.commit() self.query_one("#status-text", Label).update(f"✔ TLS Policy für {data['dom']} aktualisiert.") except Exception as e: self.query_one("#status-text", Label).update(f"❌ Fehler: {e}") conn.close() self.refresh_view() def callback_user_toggle(self, data) -> None: if not data or "@" not in data["email"]: return u, d = data["email"].split("@") conn = get_db_connection() cursor = conn.cursor() cursor.execute("UPDATE accounts SET enabled = NOT enabled WHERE username = %s AND domain = %s", (u, d)) conn.commit() conn.close() self.query_one("#status-text", Label).update(f"✔ Status für {data['email']} invertiert.") self.query_one("#menu-list", ListView).focus() def callback_alias_add(self, data) -> None: if not data or data["s_d"] == Select.BLANK or data["d_d"] == Select.BLANK: return conn = get_db_connection() cursor = conn.connector() if hasattr(conn, 'connector') else conn.cursor() try: cursor.execute("INSERT INTO aliases (source_username, source_domain, destination_username, destination_domain, enabled) VALUES (%s, %s, %s, %s, 1)", (data["s_u"], data["s_d"], data["d_u"], data["d_d"])) conn.commit() self.query_one("#status-text", Label).update("✔ Alias erfolgreich hinzugefügt.") except Exception as e: self.query_one("#status-text", Label).update(f"❌ Fehler: {e}") conn.close() self.query_one("#menu-list", ListView).focus() def callback_domain_add(self, data) -> None: if not data or not data["dom"]: return conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("INSERT INTO domains (domain) VALUES (%s)", (data["dom"].strip(),)) conn.commit() self.query_one("#status-text", Label).update(f"✔ Domain {data['dom']} hinzugefügt.") except Exception as e: self.query_one("#status-text", Label).update(f"❌ Fehler: {e}") conn.close() self.query_one("#menu-list", ListView).focus() def callback_tls_add(self, data) -> None: if not data or data["pol"] == Select.BLANK or not data["dom"]: return conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("INSERT INTO tlspolicies (domain, policy, params) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE policy=%s, params=%s", (data["dom"], data["pol"], data["param"] or None, data["pol"], data["param"] or None)) conn.commit() self.query_one("#status-text", Label).update(f"✔ TLS Policy für {data['dom']} gesetzt.") except Exception as e: self.query_one("#status-text", Label).update(f"❌ Fehler: {e}") conn.close() self.query_one("#menu-list", ListView).focus() # --- DELETE EXECUTION CALLBACKS (Nach Bestätigung) --- def callback_del_user_confirmed(self, confirmed: bool) -> None: if confirmed and hasattr(self, 'row_to_delete'): email = self.row_to_delete[1] u, d = email.split("@") conn = get_db_connection() cursor = conn.cursor() cursor.execute("DELETE FROM accounts WHERE username = %s AND domain = %s", (u, d)) conn.commit() conn.close() self.query_one("#status-text", Label).update(f"✔ Account '{email}' gelöscht.") self.refresh_view() def callback_del_alias_confirmed(self, confirmed: bool) -> None: if confirmed and hasattr(self, 'row_to_delete'): alias_id = self.row_to_delete[0] conn = get_db_connection() cursor = conn.cursor() cursor.execute("DELETE FROM aliases WHERE id = %s", (alias_id,)) conn.commit() conn.close() self.query_one("#status-text", Label).update(f"✔ Alias ID {alias_id} gelöscht.") self.refresh_view() def callback_del_domain_confirmed(self, confirmed: bool) -> None: if confirmed and hasattr(self, 'row_to_delete'): dom = self.row_to_delete[1] conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("DELETE FROM domains WHERE domain = %s", (dom,)) conn.commit() self.query_one("#status-text", Label).update(f"✔ Domain '{dom}' gelöscht.") except Exception: self.query_one("#status-text", Label).update("❌ Fehler: Abhängigkeiten (User/Aliase) blockieren das Löschen!") conn.close() self.refresh_view() def callback_del_tls_confirmed(self, confirmed: bool) -> None: if confirmed and hasattr(self, 'row_to_delete'): dom = self.row_to_delete[1] conn = get_db_connection() cursor = conn.cursor() cursor.execute("DELETE FROM tlspolicies WHERE domain = %s", (dom,)) conn.commit() conn.close() self.query_one("#status-text", Label).update(f"✔ TLS Richtlinie für '{dom}' gelöscht.") self.refresh_view() if __name__ == "__main__": VMailApp().run()