import os, re, shutil, subprocess, time from pathlib import Path SYSTEM_DATABASES = {"information_schema", "performance_schema", "mysql", "sys"} IDENT_RE = re.compile(r"^[A-Za-z0-9_$-]{1,64}$") USER_RE = re.compile(r"^[A-Za-z0-9_$.-]{1,80}$") HOST_RE = re.compile(r"^[A-Za-z0-9_.:%*\-]+$") class DatabaseProviderError(RuntimeError): pass def _client(): return shutil.which("mariadb") or shutil.which("mysql") def _dump_client(): return shutil.which("mariadb-dump") or shutil.which("mysqldump") def available(): return bool(_client()) def _run(args, timeout=60, input_text=None): try: p = subprocess.run(args, input=input_text, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout, env=os.environ.copy()) return p.returncode, p.stdout or "" except subprocess.TimeoutExpired as e: return 124, (e.stdout or "") if isinstance(e.stdout, str) else "Command timed out." except Exception as e: return 1, str(e) def _sql(sql, timeout=60): cli = _client() if not cli: raise DatabaseProviderError("MariaDB/MySQL client is not installed. Install MariaDB or MySQL first from App Store.") rc, out = _run([cli, "--batch", "--raw", "--skip-column-names", "--connect-timeout=8", "-e", sql], timeout) if rc: raise DatabaseProviderError(out.strip()[-1200:] or "Database command failed.") return out def _quote_ident(value): if not IDENT_RE.fullmatch(value or ""): raise DatabaseProviderError("Invalid database identifier. Use letters, numbers, _, -, or $.") return "`" + value.replace("`", "``") + "`" def _quote_user(value): if not USER_RE.fullmatch(value or ""): raise DatabaseProviderError("Invalid database username.") return "'" + value.replace("'", "''") + "'" def _quote_host(value): if not HOST_RE.fullmatch(value or "") or len(value) > 255: raise DatabaseProviderError("Invalid database user host.") return "'" + value.replace("'", "''") + "'" def _quote_literal(value): return "'" + str(value).replace("\\", "\\\\").replace("'", "''") + "'" def server_info(): rows = _sql("SHOW VARIABLES WHERE Variable_name IN ('hostname','port','version','version_comment')", 20) info = {} for line in rows.splitlines(): p = line.split("\t", 1) if len(p) == 2: info[p[0]] = p[1] return {"host": info.get("hostname", "localhost"), "port": int(info.get("port", "3306") or 3306), "version": info.get("version", ""), "engine": info.get("version_comment", "MariaDB/MySQL")} def list_databases(): info = server_info() raw = _sql("SHOW DATABASES", 30) names = [x.strip() for x in raw.splitlines() if x.strip() and x.strip() not in SYSTEM_DATABASES] usage = {} try: q = ("SELECT table_schema, COALESCE(SUM(data_length+index_length),0) " "FROM information_schema.tables GROUP BY table_schema") for line in _sql(q, 30).splitlines(): p = line.split("\t") if len(p) == 2: usage[p[0]] = int(p[1] or 0) except Exception: pass return [{"name": n, "size_bytes": usage.get(n, 0), "size_human": human_size(usage.get(n, 0)), "host": info["host"], "port": info["port"]} for n in names] def human_size(n): n = int(n or 0) units = ["B", "KB", "MB", "GB", "TB"] x = float(n) for u in units: if x < 1024 or u == units[-1]: return f"{x:.1f} {u}" if u != "B" else f"{int(x)} B" x /= 1024 return "0 B" def database_exists(name): name = _quote_ident(name) return bool(_sql(f"SELECT SCHEMA_NAME FROM information_schema.schemata WHERE SCHEMA_NAME={_quote_literal(name[1:-1])}", 15).strip()) def create_database(name, charset="utf8mb4", collation="utf8mb4_unicode_ci"): ident = _quote_ident(name) if not re.fullmatch(r"utf8mb4(?:_[A-Za-z0-9_]+)?", charset) or not re.fullmatch(r"utf8mb4_[A-Za-z0-9_]+", collation): raise DatabaseProviderError("Unsupported character set/collation.") _sql(f"CREATE DATABASE {ident} CHARACTER SET {charset} COLLATE {collation}", 30) def delete_database(name): _sql(f"DROP DATABASE {_quote_ident(name)}", 60) def rename_database(old_name, new_name): if old_name == new_name: raise DatabaseProviderError("Old and new database names are the same.") old = _quote_ident(old_name); new = _quote_ident(new_name) if old_name in SYSTEM_DATABASES or new_name in SYSTEM_DATABASES: raise DatabaseProviderError("System databases cannot be renamed.") if not database_exists(old_name): raise DatabaseProviderError("Source database does not exist.") if database_exists(new_name): raise DatabaseProviderError("Target database already exists.") # MySQL/MariaDB intentionally do not expose a generic RENAME DATABASE statement. # Use a real dump/import transaction-like workflow and remove the source only after import succeeds. dump = _dump_client(); cli = _client() if not dump or not cli: raise DatabaseProviderError("mysqldump/mariadb-dump and mysql/mariadb clients are required for rename.") rc, out = _run([cli, "--batch", "--raw", "--skip-column-names", "-e", f"CREATE DATABASE {new} CHARACTER SET utf8mb4"], 30) if rc: raise DatabaseProviderError(out.strip()[-1200:] or "Could not create rename target.") p = subprocess.Popen([dump, "--single-transaction", "--routines", "--events", "--triggers", "--add-drop-table", old_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False, env=os.environ.copy()) imp = subprocess.Popen([cli, new_name], stdin=p.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False, env=os.environ.copy()) if p.stdout: p.stdout.close() pout, perr = p.communicate(timeout=300) iout, ierr = imp.communicate(timeout=300) if p.returncode != 0 or imp.returncode != 0: _run([cli, "-e", f"DROP DATABASE {new}"], 30) detail = (perr or b"").decode(errors="replace")[-1000:] or (ierr or b"").decode(errors="replace")[-1000:] raise DatabaseProviderError("Database rename import failed: " + detail) _sql(f"DROP DATABASE {old}", 60) def list_users(): rows = _sql("SELECT User,Host FROM mysql.user ORDER BY User,Host", 30) out=[] for line in rows.splitlines(): p=line.split("\t") if len(p)>=2 and p[0] not in ("",): out.append({"user":p[0],"host":p[1]}) return out def create_user(user, host, password): u,h=_quote_user(user),_quote_host(host or "localhost") if not password or len(password) < 10: raise DatabaseProviderError("Password must be at least 10 characters.") # CREATE USER IF NOT EXISTS is supported by modern MariaDB/MySQL releases. _sql(f"CREATE USER IF NOT EXISTS {u}@{h} IDENTIFIED BY {_quote_literal(password)}", 30) _sql(f"ALTER USER {u}@{h} IDENTIFIED BY {_quote_literal(password)}", 30) def set_user_password(user, host, password): u,h=_quote_user(user),_quote_host(host or "localhost") if not password or len(password) < 10: raise DatabaseProviderError("Password must be at least 10 characters.") _sql(f"ALTER USER {u}@{h} IDENTIFIED BY {_quote_literal(password)}", 30) def delete_user(user, host): _sql(f"DROP USER IF EXISTS {_quote_user(user)}@{_quote_host(host or 'localhost')}", 30) def grants(user, host): rows=_sql(f"SHOW GRANTS FOR {_quote_user(user)}@{_quote_host(host or 'localhost')}", 20) return [x for x in rows.splitlines() if x.strip()] def grant_database(user, host, database, privileges): allowed={"SELECT","INSERT","UPDATE","DELETE","CREATE","DROP","ALTER","INDEX","CREATE VIEW","SHOW VIEW","TRIGGER","REFERENCES","EXECUTE","LOCK TABLES"} priv=[p.strip().upper() for p in privileges if p and p.strip().upper() in allowed] if not priv: raise DatabaseProviderError("Select at least one valid privilege.") # MySQL syntax accepts comma-separated privilege names, including multiword CREATE VIEW/SHOW VIEW. db=_quote_ident(database) _sql(f"GRANT {', '.join(priv)} ON {db}.* TO {_quote_user(user)}@{_quote_host(host or 'localhost')}", 30) def revoke_database(user, host, database, privileges=None): db=_quote_ident(database) if privileges: allowed={"SELECT","INSERT","UPDATE","DELETE","CREATE","DROP","ALTER","INDEX","CREATE VIEW","SHOW VIEW","TRIGGER","REFERENCES","EXECUTE","LOCK TABLES"} priv=[p.strip().upper() for p in privileges if p and p.strip().upper() in allowed] if not priv: raise DatabaseProviderError("Select at least one valid privilege.") sql=f"REVOKE {', '.join(priv)} ON {db}.* FROM {_quote_user(user)}@{_quote_host(host or 'localhost')}" else: sql=f"REVOKE ALL PRIVILEGES ON {db}.* FROM {_quote_user(user)}@{_quote_host(host or 'localhost')}" _sql(sql,30) def user_database_grants(user, host, database): return grants(user,host) class DatabaseProvider: name="mysql" engine="MariaDB/MySQL" @staticmethod def capabilities(): return {"create":True,"delete":True,"rename":True,"users":True,"privileges":True,"usage":True,"future_postgresql":True} create_database=staticmethod(create_database) delete_database=staticmethod(delete_database) rename_database=staticmethod(rename_database) list_databases=staticmethod(list_databases) create_user=staticmethod(create_user) set_user_password=staticmethod(set_user_password) delete_user=staticmethod(delete_user) list_users=staticmethod(list_users) grant_database=staticmethod(grant_database) revoke_database=staticmethod(revoke_database) grants=staticmethod(grants) server_info=staticmethod(server_info) class PostgreSQLProvider: """Provider contract placeholder for a future PostgreSQL backend. No PostgreSQL operation is advertised or executed in v1.3.2P1. """ name="postgresql" engine="PostgreSQL" @staticmethod def capabilities(): return {"create":False,"delete":False,"rename":False,"users":False,"privileges":False,"usage":False,"future_postgresql":True} PROVIDERS={"mysql":DatabaseProvider,"mariadb":DatabaseProvider,"postgresql":PostgreSQLProvider} def get_provider(name="mysql"): try: return PROVIDERS[name.lower()] except KeyError: raise DatabaseProviderError("Unsupported database provider.")