diff --git a/vmailtool.py b/vmailtool.py index c033867..2eb08c3 100755 --- a/vmailtool.py +++ b/vmailtool.py @@ -1,26 +1,22 @@ #!/root/sql/vmail-env/bin/python3 import mysql.connector import configparser -import getpass -import sys import socket +import sys from argon2 import PasswordHasher, Type -# Konfiguration laden +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() -DEFAULT_DOM = config.get('database', 'default_domain', fallback="") - -ph = PasswordHasher( - time_cost=3, - memory_cost=131072, - parallelism=4, - hash_len=32, - salt_len=16, - type=Type.ID -) +ph = PasswordHasher(time_cost=3, memory_cost=131072, parallelism=4, hash_len=32, salt_len=16, type=Type.ID) def get_db_connection(): try: @@ -31,192 +27,530 @@ def get_db_connection(): database=config['database']['database'] ) except Exception as e: - print(f"Fehler: {e}") + print(f"Fehler bei DB-Verbindung: {e}") sys.exit(1) def hash_pw(password): return f"{{ARGON2ID}}{ph.hash(password)}" -def get_domain_input(prompt="Domain"): - if DEFAULT_DOM: - val = input(f"{prompt} [{DEFAULT_DOM}]: ").strip() - return val if val else DEFAULT_DOM - return input(f"{prompt}: ").strip() - -# --- FUNKTIONEN --- - -def show_users(): +def fetch_all_domains(): conn = get_db_connection() cursor = conn.cursor() - cursor.execute("SELECT id, username, domain, enabled, quota FROM accounts ORDER BY id ASC") - rows = cursor.fetchall() - print(f"\n{'ID':<4} | {'Email Adresse':<35} | {'Status':<10} | {'Quota'}") - print("-" * 65) - for r in rows: - status = "AKTIV" if r[3] else "GESPERRT" - print(f"{r[0]:<4} | {r[1]+'@'+r[2]:<35} | {status:<10} | {r[4]} MB") + cursor.execute("SELECT domain FROM domains ORDER BY domain ASC") + doms = [r[0] for r in cursor.fetchall()] conn.close() + return doms -def add_user(): - user = input("Username (vor dem @): ") - dom = get_domain_input() - pw = getpass.getpass("Passwort: ") - hashed = hash_pw(pw) - conn = get_db_connection() - cursor = conn.cursor() - try: - cursor.execute("INSERT INTO accounts (username, domain, password, quota, enabled) VALUES (%s, %s, %s, 1024, 1)", - (user, dom, hashed)) - conn.commit() - print(f"✔ User {user}@{dom} angelegt.") - except Exception as e: print(f"❌ Fehler: {e}") - conn.close() -def del_user(): - email = input("E-Mail des zu LÖSCHENDEN Users: ") - if '@' not in email: return - u, d = email.split('@') - confirm = input(f"Account {email} wirklich löschen? (y/N): ") - if confirm.lower() != 'y': return - conn = get_db_connection() - cursor = conn.cursor() - cursor.execute("DELETE FROM accounts WHERE username = %s AND domain = %s", (u, d)) - conn.commit() - print("✔ User gelöscht.") - conn.close() +# --- MODAL POPUPS (Eingabemasken & Abfragen) --- -def show_aliases(): - conn = get_db_connection() - cursor = conn.cursor() - cursor.execute("SELECT id, source_username, source_domain, destination_username, destination_domain FROM aliases ORDER BY id ASC") - rows = cursor.fetchall() - print(f"\n{'ID':<4} | {'Alias (Source)':<30} | {'Ziel (Destination)':<30}") - print("-" * 70) - for r in rows: - print(f"{r[0]:<4} | {r[1]+'@'+r[2]:<30} | {r[3]+'@'+r[4]:<30}") - conn.close() +class ConfirmationModal(ModalScreen): + """Sicherheitsabfrage vor dem Löschen""" + def __init__(self, message: str): + super().__init__() + self.message = message -def add_alias(): - s_user = input("Alias Name (Source): ") - s_dom = get_domain_input("Alias Domain") - d_user = input("Ziel Username (Dest): ") - d_dom = get_domain_input("Ziel Domain") - conn = get_db_connection() - cursor = conn.cursor() - try: - cursor.execute("INSERT INTO aliases (source_username, source_domain, destination_username, destination_domain, enabled) VALUES (%s, %s, %s, %s, 1)", - (s_user, s_dom, d_user, d_dom)) - conn.commit() - print("✔ Alias angelegt.") - except Exception as e: print(f"❌ Fehler: {e}") - conn.close() + 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 del_alias(): - alias_id = input("ID des zu löschenden Alias: ") - conn = get_db_connection() - cursor = conn.cursor() - cursor.execute("DELETE FROM aliases WHERE id = %s", (alias_id,)) - conn.commit() - print(f"✔ Alias ID {alias_id} gelöscht.") - conn.close() + def on_button_pressed(self, event: Button.Pressed) -> None: + self.dismiss(event.button.id == "yes") -def show_domains(): - conn = get_db_connection() - cursor = conn.cursor() - cursor.execute("SELECT id, domain FROM domains ORDER BY id ASC") - rows = cursor.fetchall() - print(f"\n{'ID':<4} | {'Domain'}") - print("-" * 30) - for r in rows: - print(f"{r[0]:<4} | {r[1]}") - conn.close() -def add_domain(): - dom = input("Neue Domain: ").strip() - if not dom: return - conn = get_db_connection() - cursor = conn.cursor() - try: - cursor.execute("INSERT INTO domains (domain) VALUES (%s)", (dom,)) - conn.commit() - print(f"✔ Domain {dom} registriert.") - except Exception as e: print(f"❌ Fehler: {e}") - conn.close() +class EditUserModal(ModalScreen): + """Ändert Passwort und Quota zeitgleich""" + AUTO_FOCUS = "#quota" -def del_domain(): - dom = input("Zu löschende Domain (Name eingeben): ").strip() - if not dom: return - confirm = input(f"Soll die Domain '{dom}' wirklich gelöscht werden? (y/N): ") - if confirm.lower() != 'y': return + 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; } - conn = get_db_connection() - cursor = conn.cursor() - try: - cursor.execute("DELETE FROM domains WHERE domain = %s", (dom,)) + /* 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() - if cursor.rowcount > 0: - print(f"✔ Domain '{dom}' wurde gelöscht.") - else: - print("❌ Domain nicht gefunden.") - except mysql.connector.Error as e: - if e.errno == 1451: - print("❌ Fehler: Domain kann nicht gelöscht werden, da noch User oder Aliase damit verknüpft sind!") - else: - print(f"❌ Fehler: {e}") - conn.close() + conn.close() + self.query_one("#status-text", Label).update(f"✔ Status für {data['email']} invertiert.") + self.query_one("#menu-list", ListView).focus() -def change_pw(): - email = input("E-Mail für PW-Wechsel: ") - if '@' not in email: return - u, d = email.split('@') - pw = getpass.getpass("Neues Passwort: ") - hashed = hash_pw(pw) - conn = get_db_connection() - cursor = conn.cursor() - cursor.execute("UPDATE accounts SET password = %s WHERE username = %s AND domain = %s", (hashed, u, d)) - conn.commit() - print("✔ Passwort aktualisiert.") - conn.close() + 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 toggle_status(): - email = input("E-Mail zum (De)aktivieren: ") - if '@' not in email: return - u, d = 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() - print("✔ Status geändert.") - conn.close() + 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() -# --- MAIN --- + 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() -def menu(): - while True: - print("\n" + "="*60) - print(f" VMAIL MANAGER ({hostname})") - print("="*60) - print("1) Show Users 6) Del Alias") - print("2) Add User (1024MB) 7) Domains (Show/Add/Del)") - print("3) Del User 8) Change Password") - print("4) Show Aliases 9) Toggle Status (An/Aus)") - print("5) Add Alias 0) Beenden") - - cmd = input("\nAuswahl > ") - if cmd == '1': show_users() - elif cmd == '2': add_user() - elif cmd == '3': del_user() - elif cmd == '4': show_aliases() - elif cmd == '5': add_alias() - elif cmd == '6': del_alias() - elif cmd == '7': - show_domains() - sub_cmd = input("\n[A]dd Domain, [D]el Domain oder [Enter] zurück: ").lower() - if sub_cmd == 'a': add_domain() - elif sub_cmd == 'd': del_domain() - elif cmd == '8': change_pw() - elif cmd == '9': toggle_status() - elif cmd == '0': break if __name__ == "__main__": - menu() + VMailApp().run()