-- 创建用户 CREATE USER 'app_user'@'localhost' IDENTIFIED BY 'secure_password'; -- 创建带过期时间的用户 CREATE USER 'temp_user'@'%' IDENTIFIED BY 'password' WITH EXPIRE INTERVAL 30 DAY; -- 修改用户密码 ALTER USER 'app_user'@'localhost' IDENTIFIED BY 'new_password'; -- 锁定用户 ALTER USER 'suspicious_user'@'localhost' ACCOUNT LOCK; -- 删除用户 DROP USER 'old_user'@'localhost';
2.2 权限管理
-- 授予数据库权限 GRANT SELECT, INSERT, UPDATE ON mydatabase.* TO 'app_user'@'localhost'; -- 授予表级权限 GRANT SELECT ON mydatabase.users TO 'read_only_user'@'%'; -- 授予列级权限 GRANT SELECT (id, name) ON mydatabase.users TO 'limited_user'@'localhost'; -- 授予管理权限 GRANT ALL PRIVILEGES ON *.* TO 'admin'@'localhost' WITH GRANT OPTION; -- 撤销权限 REVOKE DELETE ON mydatabase.orders FROM 'app_user'@'localhost'; -- 查看权限 SHOW GRANTS FOR 'app_user'@'localhost';
2.3 安全工具
class DatabaseSecurityManager: def __init__(self, connection): self.connection = connection def create_user(self, username, password, host='localhost', privileges=None): cursor = self.connection.cursor() cursor.execute(f"CREATE USER '{username}'@'{host}' IDENTIFIED BY '{password}'") if privileges: for db, privs in privileges.items(): priv_str = ', '.join(privs) cursor.execute(f"GRANT {priv_str} ON {db} TO '{username}'@'{host}'") self.connection.commit() def audit_user_actions(self, username=None): cursor = self.connection.cursor() if username: query = f"SELECT * FROM audit_log WHERE user = '{username}'" else: query = "SELECT * FROM audit_log ORDER BY timestamp DESC LIMIT 100" cursor.execute(query) return cursor.fetchall() def check_weak_passwords(self): weak_patterns = ['password', '123456', 'qwerty'] cursor = self.connection.cursor() cursor.execute("SELECT user, host FROM mysql.user") weak_users = [] for user, host in cursor.fetchall(): cursor.execute(f"SELECT authentication_string FROM mysql.user WHERE user = '{user}' AND host = '{host}'") hash_val = cursor.fetchone()[0] for pattern in weak_patterns: if pattern.lower() in user.lower(): weak_users.append(user) break return weak_users def validate_permissions(self): cursor = self.connection.cursor() cursor.execute(""" SELECT user, host, privilege_type, table_name FROM INFORMATION_SCHEMA.USER_PRIVILEGES WHERE privilege_type = 'ALL' """) overprivileged = cursor.fetchall() return overprivileged