""" User & Privilege Management panel. Supports MySQL and PostgreSQL. SQLite and MSSQL show an informational message (no user management via this UI). Features -------- • List all database users with host (MySQL) or connection limit (PostgreSQL) • Create user (username + password + optional host for MySQL) • Drop user with confirmation • View per-database GRANT privileges for the selected user • GRANT / REVOKE individual privileges on a database All DDL operations execute synchronously in a background QThread to keep the UI responsive. """ from __future__ import annotations from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QSplitter, QTableWidget, QTableWidgetItem, QHeaderView, QLabel, QPushButton, QComboBox, QCheckBox, QDialog, QFormLayout, QLineEdit, QDialogButtonBox, QMessageBox, QGroupBox, QAbstractItemView, ) from PyQt6.QtCore import Qt, QThread, pyqtSignal from PyQt6.QtGui import QColor, QFont # ── Standard SQL privileges offered in the GRANT UI ────────────────────────── _MYSQL_PRIVS = [ "SELECT", "INSERT", "UPDATE", "DELETE", "CREATE", "DROP", "ALTER", "INDEX", "CREATE VIEW", "SHOW VIEW", "CREATE ROUTINE", "ALTER ROUTINE", "EXECUTE", "REFERENCES", "TRIGGER", "LOCK TABLES", "CREATE TEMPORARY TABLES", ] _PG_PRIVS = [ "SELECT", "INSERT", "UPDATE", "DELETE", "TRUNCATE", "REFERENCES", "TRIGGER", "CREATE", "CONNECT", "TEMPORARY", "EXECUTE", "USAGE", ] # ── Background worker ───────────────────────────────────────────────────────── class _UserWorker(QThread): """Run an arbitrary callable in a background thread.""" result = pyqtSignal(object) error = pyqtSignal(str) def __init__(self, fn, *args, parent=None): super().__init__(parent) self._fn = fn self._args = args def run(self): try: self.result.emit(self._fn(*self._args)) except Exception as e: self.error.emit(str(e)) # ── Driver-level helpers (attached at runtime, no driver subclassing needed) ── def _mysql_get_users(driver) -> list[dict]: """Return list of {user, host, super} from MySQL.user table.""" cols, rows, _ = driver.execute_query( "SELECT User, Host, Super_priv FROM mysql.user ORDER BY User, Host" ) return [{"user": r[0], "host": r[1], "super": r[2]} for r in rows] def _mysql_get_grants(driver, user: str, host: str) -> list[str]: cols, rows, _ = driver.execute_query( f"SHOW GRANTS FOR %s@%s", (user, host) # type: ignore[arg-type] ) return [r[0] for r in rows] def _mysql_create_user(driver, user: str, host: str, password: str): driver.execute_query( "CREATE USER %s@%s IDENTIFIED BY %s", (user, host, password) ) def _mysql_drop_user(driver, user: str, host: str): driver.execute_query(f"DROP USER %s@%s", (user, host)) def _mysql_grant(driver, privs: list[str], database: str, user: str, host: str): priv_str = ", ".join(privs) if privs else "USAGE" driver.execute_query( f"GRANT {priv_str} ON `{database}`.* TO %s@%s", (user, host) ) driver.execute_query("FLUSH PRIVILEGES") def _mysql_revoke(driver, privs: list[str], database: str, user: str, host: str): priv_str = ", ".join(privs) if privs else "USAGE" driver.execute_query( f"REVOKE {priv_str} ON `{database}`.* FROM %s@%s", (user, host) ) driver.execute_query("FLUSH PRIVILEGES") # ── PostgreSQL helpers ──────────────────────────────────────────────────────── def _pg_get_users(driver) -> list[dict]: cols, rows, _ = driver.execute_query( "SELECT usename, usesuper, usecreatedb FROM pg_user ORDER BY usename" ) return [{"user": r[0], "host": "", "super": "Y" if r[1] else "N", "createdb": "Y" if r[2] else "N"} for r in rows] def _pg_get_grants(driver, user: str, _host: str) -> list[str]: cols, rows, _ = driver.execute_query(""" SELECT grantee, table_catalog, table_schema, table_name, string_agg(privilege_type, ', ') AS privileges FROM information_schema.role_table_grants WHERE grantee = %s GROUP BY grantee, table_catalog, table_schema, table_name ORDER BY table_catalog, table_schema, table_name """, (user,)) return [ f"GRANT {r[4]} ON {r[1]}.{r[2]}.{r[3]} TO {r[0]}" for r in rows ] def _pg_create_user(driver, user: str, _host: str, password: str): driver.execute_query( f"CREATE USER \"{user}\" WITH PASSWORD %s", (password,) ) def _pg_drop_user(driver, user: str, _host: str): driver.execute_query(f'DROP USER "{user}"') def _pg_grant(driver, privs: list[str], database: str, user: str, _host: str): priv_str = ", ".join(privs) if privs else "CONNECT" driver.execute_query( f'GRANT {priv_str} ON DATABASE "{database}" TO "{user}"' ) def _pg_revoke(driver, privs: list[str], database: str, user: str, _host: str): priv_str = ", ".join(privs) if privs else "CONNECT" driver.execute_query( f'REVOKE {priv_str} ON DATABASE "{database}" FROM "{user}"' ) # ── Dialogs ─────────────────────────────────────────────────────────────────── class _CreateUserDialog(QDialog): def __init__(self, db_type: str, parent=None): super().__init__(parent) self.setWindowTitle("Create User") self.setModal(True) self.setMinimumWidth(360) self._db_type = db_type root = QVBoxLayout(self) form = QFormLayout() form.setSpacing(8) self._user_edit = QLineEdit() self._user_edit.setPlaceholderText("e.g. app_user") form.addRow("Username:", self._user_edit) if db_type == "mysql": self._host_edit = QLineEdit() self._host_edit.setText("%") self._host_edit.setPlaceholderText("% = any host") form.addRow("Host:", self._host_edit) self._pass_edit = QLineEdit() self._pass_edit.setEchoMode(QLineEdit.EchoMode.Password) self._pass_edit.setPlaceholderText("Password") form.addRow("Password:", self._pass_edit) self._pass2_edit = QLineEdit() self._pass2_edit.setEchoMode(QLineEdit.EchoMode.Password) self._pass2_edit.setPlaceholderText("Confirm password") form.addRow("Confirm:", self._pass2_edit) root.addLayout(form) bbox = QDialogButtonBox( QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel, ) bbox.accepted.connect(self._validate) bbox.rejected.connect(self.reject) root.addWidget(bbox) def _validate(self): if not self._user_edit.text().strip(): QMessageBox.warning(self, "Validation", "Username is required.") return if self._pass_edit.text() != self._pass2_edit.text(): QMessageBox.warning(self, "Validation", "Passwords do not match.") return self.accept() @property def username(self) -> str: return self._user_edit.text().strip() @property def host(self) -> str: if self._db_type == "mysql": return self._host_edit.text().strip() or "%" return "" @property def password(self) -> str: return self._pass_edit.text() class _GrantDialog(QDialog): """Select privileges + target database and grant/revoke.""" def __init__(self, driver, db_type: str, databases: list[str], user: str, host: str, mode: str = "grant", parent=None): super().__init__(parent) self._driver = driver self._db_type = db_type self._user = user self._host = host self._mode = mode # "grant" | "revoke" self.setWindowTitle(f"{'Grant to' if mode == 'grant' else 'Revoke from'} {user}") self.setModal(True) self.setMinimumSize(420, 480) self._build_ui(databases) def _build_ui(self, databases: list[str]): root = QVBoxLayout(self) form = QFormLayout() self._db_combo = QComboBox() for db in databases: self._db_combo.addItem(db) form.addRow("Database:", self._db_combo) root.addLayout(form) # Privilege checkboxes priv_box = QGroupBox("Privileges") pb_lay = QVBoxLayout(priv_box) privs = _MYSQL_PRIVS if self._db_type == "mysql" else _PG_PRIVS self._priv_checks: list[QCheckBox] = [] for priv in privs: cb = QCheckBox(priv) pb_lay.addWidget(cb) self._priv_checks.append(cb) sel_row = QHBoxLayout() sel_all = QPushButton("All") sel_all.setFixedWidth(50) sel_all.clicked.connect(lambda: [cb.setChecked(True) for cb in self._priv_checks]) sel_none = QPushButton("None") sel_none.setFixedWidth(50) sel_none.clicked.connect(lambda: [cb.setChecked(False) for cb in self._priv_checks]) sel_row.addWidget(sel_all) sel_row.addWidget(sel_none) sel_row.addStretch() pb_lay.insertLayout(0, sel_row) root.addWidget(priv_box, 1) verb = "Grant" if self._mode == "grant" else "Revoke" bbox = QDialogButtonBox() ok = bbox.addButton(verb, QDialogButtonBox.ButtonRole.AcceptRole) ok.clicked.connect(self._execute) bbox.addButton(QDialogButtonBox.StandardButton.Cancel).clicked.connect( self.reject) root.addWidget(bbox) def _execute(self): privs = [cb.text() for cb in self._priv_checks if cb.isChecked()] if not privs: QMessageBox.warning(self, "No Privilege", "Select at least one privilege.") return db = self._db_combo.currentText() try: if self._db_type == "mysql": fn = _mysql_grant if self._mode == "grant" else _mysql_revoke else: fn = _pg_grant if self._mode == "grant" else _pg_revoke fn(self._driver, privs, db, self._user, self._host) self.accept() except Exception as e: QMessageBox.critical(self, "Error", str(e)) @property def selected_privs(self) -> list[str]: return [cb.text() for cb in self._priv_checks if cb.isChecked()] # ── Main Panel ──────────────────────────────────────────────────────────────── class UserManagerPanel(QWidget): """ User & privilege management workspace tab. Parameters ---------- driver : BaseDriver — must already be connected parent : QWidget """ status_message = pyqtSignal(str) def __init__(self, driver, parent=None): super().__init__(parent) self._driver = driver self._db_type = getattr(driver, "db_type", "").lower() self._databases: list[str] = [] self._current_user: str = "" self._current_host: str = "" self._worker: _UserWorker | None = None self._build_ui() self._load_users() self._load_databases() # ── UI construction ─────────────────────────────────────────────────────── def _build_ui(self): root = QVBoxLayout(self) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(0) # ── Header bar ──────────────────────────────────────────────────────── hdr = QHBoxLayout() hdr.setContentsMargins(8, 6, 8, 6) hdr.setSpacing(6) title = QLabel("👤 User & Privilege Management") f = title.font() f.setBold(True) f.setPointSize(11) title.setFont(f) title.setObjectName("structureTitle") self._refresh_btn = QPushButton("🔄 Refresh") self._refresh_btn.setFixedWidth(90) self._refresh_btn.clicked.connect(self._load_users) hdr.addWidget(title) hdr.addStretch() hdr.addWidget(self._refresh_btn) root.addLayout(hdr) # ── Unsupported banner (hidden for MySQL/PG) ────────────────────────── self._unsupported_lbl = QLabel( "ℹ️ User management is not available for this database type.\n" " (Supports MySQL and PostgreSQL only.)" ) self._unsupported_lbl.setAlignment(Qt.AlignmentFlag.AlignCenter) self._unsupported_lbl.setObjectName("emptyLabel") root.addWidget(self._unsupported_lbl) if self._db_type not in ("mysql", "postgresql"): self._unsupported_lbl.setVisible(True) return self._unsupported_lbl.setVisible(False) # ── Main splitter: user list (left) / grants detail (right) ────────── splitter = QSplitter(Qt.Orientation.Horizontal) # ── Left: user table ────────────────────────────────────────────────── left = QWidget() ll = QVBoxLayout(left) ll.setContentsMargins(0, 0, 0, 0) ll.addWidget(self._make_section_label("Users")) # Toolbar tb = QHBoxLayout() tb.setContentsMargins(4, 0, 4, 4) self._add_user_btn = QPushButton("➕ Create User") self._drop_user_btn = QPushButton("🗑️ Drop User") self._drop_user_btn.setObjectName("crudDeleteBtn") self._drop_user_btn.setEnabled(False) self._add_user_btn.clicked.connect(self._create_user) self._drop_user_btn.clicked.connect(self._drop_user) tb.addWidget(self._add_user_btn) tb.addWidget(self._drop_user_btn) tb.addStretch() ll.addLayout(tb) self._user_table = QTableWidget() self._user_table.setAlternatingRowColors(True) self._user_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers) self._user_table.setSelectionBehavior( QAbstractItemView.SelectionBehavior.SelectRows) self._user_table.setSelectionMode( QAbstractItemView.SelectionMode.SingleSelection) self._user_table.verticalHeader().setDefaultSectionSize(24) self._user_table.itemSelectionChanged.connect(self._on_user_selected) ll.addWidget(self._user_table) splitter.addWidget(left) # ── Right: grants panel ─────────────────────────────────────────────── right = QWidget() rl = QVBoxLayout(right) rl.setContentsMargins(0, 0, 0, 0) rl.addWidget(self._make_section_label("Grants for selected user")) # Grant toolbar gtb = QHBoxLayout() gtb.setContentsMargins(4, 0, 4, 4) self._grant_btn = QPushButton("＋ Grant") self._revoke_btn = QPushButton("－ Revoke") self._grant_btn.setEnabled(False) self._revoke_btn.setEnabled(False) self._grant_btn.clicked.connect(self._grant_privs) self._revoke_btn.clicked.connect(self._revoke_privs) gtb.addWidget(self._grant_btn) gtb.addWidget(self._revoke_btn) gtb.addStretch() rl.addLayout(gtb) self._grants_table = QTableWidget(0, 1) self._grants_table.setHorizontalHeaderLabels(["GRANT Statement"]) self._grants_table.horizontalHeader().setSectionResizeMode( QHeaderView.ResizeMode.Stretch) self._grants_table.setAlternatingRowColors(True) self._grants_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers) self._grants_table.verticalHeader().setDefaultSectionSize(24) rl.addWidget(self._grants_table) splitter.addWidget(right) splitter.setSizes([340, 560]) root.addWidget(splitter, 1) # ── Status bar ──────────────────────────────────────────────────────── self._status_lbl = QLabel("") self._status_lbl.setObjectName("rowCountLbl") self._status_lbl.setContentsMargins(8, 4, 0, 4) root.addWidget(self._status_lbl) @staticmethod def _make_section_label(text: str) -> QLabel: lbl = QLabel(f" {text}") lbl.setObjectName("sidebarTitle") lbl.setContentsMargins(0, 4, 0, 4) f = lbl.font() f.setBold(True) lbl.setFont(f) return lbl # ── Data loading ────────────────────────────────────────────────────────── def _load_users(self): if self._db_type not in ("mysql", "postgresql"): return fn = _mysql_get_users if self._db_type == "mysql" else _pg_get_users self._worker = _UserWorker(fn, self._driver, parent=self) self._worker.result.connect(self._populate_users) self._worker.error.connect( lambda e: self._set_status(f"Error loading users: {e}")) self._worker.start() def _load_databases(self): try: self._databases = self._driver.get_databases() except Exception: self._databases = [] def _populate_users(self, users: list[dict]): t = self._user_table t.clear() if self._db_type == "mysql": t.setColumnCount(3) t.setHorizontalHeaderLabels(["Username", "Host", "Super"]) else: t.setColumnCount(3) t.setHorizontalHeaderLabels(["Username", "Superuser", "CreateDB"]) t.setRowCount(len(users)) t.horizontalHeader().setSectionResizeMode( QHeaderView.ResizeMode.ResizeToContents) t.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch) for r, u in enumerate(users): if self._db_type == "mysql": vals = [u["user"], u["host"], u.get("super", "")] else: vals = [u["user"], u.get("super", ""), u.get("createdb", "")] for c, v in enumerate(vals): item = QTableWidgetItem(str(v)) item.setTextAlignment(Qt.AlignmentFlag.AlignCenter) t.setItem(r, c, item) self._set_status(f"{len(users)} user(s)") def _on_user_selected(self): rows = self._user_table.selectionModel().selectedRows() if not rows: self._drop_user_btn.setEnabled(False) self._grant_btn.setEnabled(False) self._revoke_btn.setEnabled(False) return r = rows[0].row() self._current_user = self._user_table.item(r, 0).text() self._current_host = ( self._user_table.item(r, 1).text() if self._db_type == "mysql" else "" ) self._drop_user_btn.setEnabled(True) self._grant_btn.setEnabled(True) self._revoke_btn.setEnabled(True) self._load_grants() def _load_grants(self): if not self._current_user: return fn = _mysql_get_grants if self._db_type == "mysql" else _pg_get_grants self._worker = _UserWorker( fn, self._driver, self._current_user, self._current_host, parent=self ) self._worker.result.connect(self._populate_grants) self._worker.error.connect( lambda e: self._set_status(f"Error loading grants: {e}")) self._worker.start() def _populate_grants(self, grants: list[str]): t = self._grants_table t.setRowCount(len(grants)) mono = QFont("Consolas", 10) for r, g in enumerate(grants): item = QTableWidgetItem(g) item.setFont(mono) t.setItem(r, 0, item) self._set_status( f"{len(grants)} grant statement(s) for {self._current_user}" ) # ── CRUD operations ─────────────────────────────────────────────────────── def _create_user(self): dlg = _CreateUserDialog(self._db_type, parent=self) if not dlg.exec(): return fn = _mysql_create_user if self._db_type == "mysql" else _pg_create_user self._worker = _UserWorker( fn, self._driver, dlg.username, dlg.host, dlg.password, parent=self ) self._worker.result.connect(lambda _: ( self._set_status(f"User '{dlg.username}' created."), self._load_users(), )) self._worker.error.connect( lambda e: QMessageBox.critical(self, "Create User Error", str(e)) ) self._worker.start() def _drop_user(self): if not self._current_user: return btn = QMessageBox.warning( self, "Drop User", f"Drop user '{self._current_user}' permanently?\n\n" "All associated privileges will be revoked.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No, ) if btn != QMessageBox.StandardButton.Yes: return fn = _mysql_drop_user if self._db_type == "mysql" else _pg_drop_user self._worker = _UserWorker( fn, self._driver, self._current_user, self._current_host, parent=self ) self._worker.result.connect(lambda _: ( self._set_status(f"User '{self._current_user}' dropped."), self._load_users(), )) self._worker.error.connect( lambda e: QMessageBox.critical(self, "Drop User Error", str(e)) ) self._worker.start() def _grant_privs(self): dlg = _GrantDialog( self._driver, self._db_type, self._databases, self._current_user, self._current_host, mode="grant", parent=self, ) if dlg.exec(): self._set_status( f"Privileges granted to {self._current_user}." ) self._load_grants() def _revoke_privs(self): dlg = _GrantDialog( self._driver, self._db_type, self._databases, self._current_user, self._current_host, mode="revoke", parent=self, ) if dlg.exec(): self._set_status( f"Privileges revoked from {self._current_user}." ) self._load_grants() # ── Helpers ─────────────────────────────────────────────────────────────── def _set_status(self, msg: str): self._status_lbl.setText(msg) self.status_message.emit(msg)