Python脚本如何操作数据库用户权限

本文目录导读：

1. MySQL数据库权限操作
2. PostgreSQL数据库权限操作
3. 高级功能：权限管理工具类
4. 安全建议
5. 注意事项

我来详细介绍Python操作数据库用户权限的方法,主要针对MySQL和PostgreSQL两种常用数据库。

MySQL数据库权限操作

安装依赖

pip install pymysql
# 或
pip install mysql-connector-python

基本权限操作示例

import pymysql
from pymysql.cursors import DictCursor
class MySQLPrivilegeManager:
    def __init__(self, host='localhost', user='root', password='root_password', port=3306):
        """初始化数据库连接（需要管理员权限）"""
        self.connection = pymysql.connect(
            host=host,
            user=user,
            password=password,
            port=port,
            cursorclass=DictCursor
        )
        self.cursor = self.connection.cursor()
    def create_user(self, username, password, host='localhost'):
        """创建新用户"""
        try:
            sql = f"CREATE USER '{username}'@'{host}' IDENTIFIED BY '{password}'"
            self.cursor.execute(sql)
            self.connection.commit()
            return f"用户 {username} 创建成功"
        except Exception as e:
            return f"创建用户失败: {str(e)}"
    def grant_privileges(self, username, database, table='*', privileges=['SELECT','INSERT','UPDATE'], host='localhost'):
        """授予权限"""
        try:
            priv_str = ', '.join(privileges)
            sql = f"GRANT {priv_str} ON {database}.{table} TO '{username}'@'{host}'"
            self.cursor.execute(sql)
            self.cursor.execute("FLUSH PRIVILEGES")
            self.connection.commit()
            return f"权限授予成功: {priv_str} ON {database}.{table}"
        except Exception as e:
            return f"权限授予失败: {str(e)}"
    def revoke_privileges(self, username, database, table='*', privileges=['ALL'], host='localhost'):
        """撤销权限"""
        try:
            priv_str = ', '.join(privileges)
            sql = f"REVOKE {priv_str} ON {database}.{table} FROM '{username}'@'{host}'"
            self.cursor.execute(sql)
            self.cursor.execute("FLUSH PRIVILEGES")
            self.connection.commit()
            return f"权限撤销成功"
        except Exception as e:
            return f"权限撤销失败: {str(e)}"
    def show_user_privileges(self, username, host='localhost'):
        """查看用户权限"""
        try:
            sql = f"SHOW GRANTS FOR '{username}'@'{host}'"
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            return [list(result.values())[0] for result in results]
        except Exception as e:
            return f"查询失败: {str(e)}"
    def drop_user(self, username, host='localhost'):
        """删除用户"""
        try:
            sql = f"DROP USER '{username}'@'{host}'"
            self.cursor.execute(sql)
            self.connection.commit()
            return f"用户 {username} 已删除"
        except Exception as e:
            return f"删除用户失败: {str(e)}"
    def close(self):
        """关闭连接"""
        self.cursor.close()
        self.connection.close()
# 使用示例
if __name__ == "__main__":
    # 使用管理员账号连接
    pm = MySQLPrivilegeManager(
        host='localhost',
        user='root',
        password='your_root_password'
    )
    # 创建用户
    print(pm.create_user('app_user', 'secure_password123'))
    # 授予权限
    print(pm.grant_privileges(
        username='app_user',
        database='test_db',
        privileges=['SELECT', 'INSERT', 'UPDATE', 'DELETE']
    ))
    # 查看权限
    privileges = pm.show_user_privileges('app_user')
    for priv in privileges:
        print(priv)
    # 撤销权限
    print(pm.revoke_privileges('app_user', 'test_db', privileges=['DELETE']))
    # 删除用户
    print(pm.drop_user('app_user'))
    pm.close()

PostgreSQL数据库权限操作

安装依赖

pip install psycopg2-binary

PostgreSQL权限操作示例

import psycopg2
from psycopg2.extras import RealDictCursor
class PostgresPrivilegeManager:
    def __init__(self, host='localhost', user='postgres', password='postgres', port=5432):
        """初始化数据库连接（需要管理员或超级用户）"""
        self.connection = psycopg2.connect(
            host=host,
            user=user,
            password=password,
            port=port
        )
        self.connection.autocommit = True
        self.cursor = self.connection.cursor()
    def create_user(self, username, password, login=True):
        """创建用户（角色）"""
        try:
            login_str = "LOGIN" if login else "NOLOGIN"
            sql = f"CREATE ROLE {username} WITH {login_str} PASSWORD '{password}'"
            self.cursor.execute(sql)
            return f"用户 {username} 创建成功"
        except Exception as e:
            return f"创建用户失败: {str(e)}"
    def grant_database_privileges(self, username, database, privileges=['CONNECT','CREATE']):
        """授予数据库级别权限"""
        try:
            for privilege in privileges:
                sql = f"GRANT {privilege} ON DATABASE {database} TO {username}"
                self.cursor.execute(sql)
            return f"数据库权限授予成功: {', '.join(privileges)} ON {database}"
        except Exception as e:
            return f"权限授予失败: {str(e)}"
    def grant_schema_privileges(self, username, schema='public', privileges=['USAGE','CREATE']):
        """授予Schema权限"""
        try:
            for privilege in privileges:
                sql = f"GRANT {privilege} ON SCHEMA {schema} TO {username}"
                self.cursor.execute(sql)
            return f"Schema权限授予成功"
        except Exception as e:
            return f"权限授予失败: {str(e)}"
    def grant_table_privileges(self, username, table_name, privileges=['SELECT','INSERT','UPDATE']):
        """授予表级别权限"""
        try:
            priv_str = ', '.join(privileges)
            sql = f"GRANT {priv_str} ON TABLE {table_name} TO {username}"
            self.cursor.execute(sql)
            return f"表权限授予成功"
        except Exception as e:
            return f"权限授予失败: {str(e)}"
    def grant_all_tables_in_schema(self, username, schema='public', privileges=['SELECT','INSERT','UPDATE','DELETE']):
        """授予Schema中所有表权限"""
        try:
            # 设置默认权限
            priv_str = ', '.join(privileges)
            sql = f"ALTER DEFAULT PRIVILEGES IN SCHEMA {schema} GRANT {priv_str} ON TABLES TO {username}"
            self.cursor.execute(sql)
            # 授予现有表权限
            sql = f"GRANT {priv_str} ON ALL TABLES IN SCHEMA {schema} TO {username}"
            self.cursor.execute(sql)
            return f"Schema中所有表权限授予成功"
        except Exception as e:
            return f"权限授予失败: {str(e)}"
    def revoke_privileges(self, username, object_type, object_name, privileges=['ALL']):
        """撤销权限"""
        try:
            priv_str = ', '.join(privileges)
            sql = f"REVOKE {priv_str} ON {object_type} {object_name} FROM {username}"
            self.cursor.execute(sql)
            return f"权限撤销成功"
        except Exception as e:
            return f"权限撤销失败: {str(e)}"
    def show_user_privileges(self, username):
        """查看用户权限"""
        try:
            # 查询数据库权限
            self.cursor.execute("""
                SELECT datname, datacl 
                FROM pg_database 
                WHERE datname IN (SELECT datname FROM pg_database)
            """)
            # 更详细的查询方法
            sql = f"""
                SELECT 
                    grantee,
                    table_schema,
                    table_name,
                    privilege_type
                FROM information_schema.role_table_grants
                WHERE grantee = '{username}'
            """
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            return results
        except Exception as e:
            return f"查询失败: {str(e)}"
    def drop_user(self, username):
        """删除用户"""
        try:
            # 先转移对象所有权
            sql = f"REASSIGN OWNED BY {username} TO postgres"
            self.cursor.execute(sql)
            # 删除用户
            sql = f"DROP ROLE IF EXISTS {username}"
            self.cursor.execute(sql)
            return f"用户 {username} 已删除"
        except Exception as e:
            return f"删除用户失败: {str(e)}"
    def close(self):
        """关闭连接"""
        self.cursor.close()
        self.connection.close()
# 使用示例
if __name__ == "__main__":
    # 连接PostgreSQL
    pm = PostgresPrivilegeManager(
        host='localhost',
        user='postgres',
        password='your_postgres_password'
    )
    # 创建用户
    print(pm.create_user('app_user', 'secure_pass_123'))
    # 授予数据库权限
    print(pm.grant_database_privileges('app_user', 'test_db'))
    # 授予Schema权限
    print(pm.grant_schema_privileges('app_user'))
    # 授予特定表权限
    print(pm.grant_table_privileges('app_user', 'users'))
    # 授予Schema中所有表权限
    print(pm.grant_all_tables_in_schema('app_user', 'public'))
    # 查看权限
    privileges = pm.show_user_privileges('app_user')
    print("用户权限:", privileges)
    # 撤销权限
    print(pm.revoke_privileges('app_user', 'TABLE', 'users'))
    # 删除用户
    print(pm.drop_user('app_user'))
    pm.close()

高级功能：权限管理工具类

import json
import yaml
class DatabasePrivilegeManager:
    """统一的数据库权限管理工具"""
    def __init__(self, db_type='mysql', **connection_params):
        self.db_type = db_type
        if db_type == 'mysql':
            self.manager = MySQLPrivilegeManager(**connection_params)
        elif db_type == 'postgresql':
            self.manager = PostgresPrivilegeManager(**connection_params)
        else:
            raise ValueError(f"不支持的数据库类型: {db_type}")
    def batch_create_users(self, users_config):
        """批量创建用户"""
        results = []
        for user in users_config:
            result = self.manager.create_user(
                user['username'],
                user['password'],
                user.get('host', 'localhost')
            )
            results.append(result)
        return results
    def load_permissions_from_file(self, config_file):
        """从配置文件加载权限设置"""
        if config_file.endswith('.json'):
            with open(config_file, 'r') as f:
                config = json.load(f)
        elif config_file.endswith('.yaml') or config_file.endswith('.yml'):
            with open(config_file, 'r') as f:
                config = yaml.safe_load(f)
        else:
            raise ValueError("不支持的配置文件格式")
        return config
    def apply_permissions_from_config(self, config):
        """根据配置应用权限"""
        results = []
        # 处理用户和权限
        for user_config in config.get('users', []):
            username = user_config['username']
            password = user_config['password']
            # 创建用户
            result = self.manager.create_user(username, password)
            results.append(result)
            # 授予数据库权限
            for db_perm in user_config.get('database_privileges', []):
                result = self.manager.grant_database_privileges(
                    username, 
                    db_perm['database'],
                    db_perm.get('privileges', ['CONNECT', 'CREATE'])
                )
                results.append(result)
            # 授予表权限
            for table_perm in user_config.get('table_privileges', []):
                if self.db_type == 'postgresql':
                    result = self.manager.grant_table_privileges(
                        username,
                        f"{table_perm['schema']}.{table_perm['table']}",
                        table_perm.get('privileges', ['SELECT', 'INSERT', 'UPDATE'])
                    )
                else:
                    result = self.manager.grant_privileges(
                        username,
                        table_perm['database'],
                        table_perm['table'],
                        table_perm.get('privileges', ['SELECT', 'INSERT', 'UPDATE'])
                    )
                results.append(result)
        return results
    def generate_audit_report(self):
        """生成权限审计报告"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'database_type': self.db_type,
            'users': {}
        }
        # 查询用户列表
        if self.db_type == 'mysql':
            self.manager.cursor.execute("SELECT User, Host FROM mysql.user")
            users = self.manager.cursor.fetchall()
            for user in users:
                username = user['User']
                host = user['Host']
                privileges = self.manager.show_user_privileges(username, host)
                report['users'][f"{username}@{host}"] = privileges
        else:
            self.manager.cursor.execute("SELECT rolname FROM pg_roles")
            users = self.manager.cursor.fetchall()
            for user in users:
                username = user[0]
                privileges = self.manager.show_user_privileges(username)
                report['users'][username] = privileges
        return report
    def export_config(self, file_path):
        """导出当前权限配置"""
        report = self.generate_audit_report()
        # 转换为可加载的配置格式
        config = {
            'users': []
        }
        for username, privileges in report['users'].items():
            user_config = {
                'username': username,
                'privileges': privileges
            }
            config['users'].append(user_config)
        if file_path.endswith('.json'):
            with open(file_path, 'w') as f:
                json.dump(config, f, indent=2, ensure_ascii=False)
        elif file_path.endswith('.yaml'):
            with open(file_path, 'w') as f:
                yaml.dump(config, f, default_flow_style=False)
        return f"配置已导出到 {file_path}"
# 使用高级功能
def advanced_usage():
    # 初始化管理器
    pm = DatabasePrivilegeManager(
        db_type='mysql',
        host='localhost',
        user='root',
        password='root_password'
    )
    # 从YAML配置文件加载权限
    config = pm.load_permissions_from_file('permissions.yaml')
    # YAML配置示例:
    # users:
    #   - username: app_user
    #     password: secure_pass
    #     database_privileges:
    #       - database: app_db
    #         privileges: [CONNECT, CREATE]
    #     table_privileges:
    #       - database: app_db
    #         table: users
    #         privileges: [SELECT, INSERT, UPDATE]
    # 应用配置
    results = pm.apply_permissions_from_config(config)
    for result in results:
        print(result)
    # 生成审计报告
    report = pm.generate_audit_report()
    print(json.dumps(report, indent=2, default=str))
    # 导出配置
    pm.export_config('permissions_backup.json')
if __name__ == "__main__":
    advanced_usage()

安全建议

class SecurePrivilegeManager:
    """安全增强的权限管理器"""
    def __init__(self, **connection_params):
        # 使用安全连接
        self.connection_params = {
            'host': connection_params.get('host', 'localhost'),
            'port': connection_params.get('port', 3306),
            'ssl_ca': connection_params.get('ssl_ca'),
            'ssl_cert': connection_params.get('ssl_cert'),
            'ssl_key': connection_params.get('ssl_key')
        }
        # 密码加密存储
        self.password = connection_params.get('password')
    def create_user_with_limits(self, username, password, max_queries_per_hour=0, max_updates_per_hour=0, max_connections_per_hour=0):
        """创建用户并设置资源限制"""
        sql = f"""
            CREATE USER '{username}'@'%' 
            IDENTIFIED BY '{password}'
            WITH 
                MAX_QUERIES_PER_HOUR {max_queries_per_hour}
                MAX_UPDATES_PER_HOUR {max_updates_per_hour}
                MAX_CONNECTIONS_PER_HOUR {max_connections_per_hour}
        """
        # 执行SQL...
        pass
    def grant_minimal_privileges(self, username, database):
        """授予最小必需权限"""
        # 只授予必要的权限
        minimal_privileges = ['SELECT', 'INSERT', 'UPDATE']
        return self.grant_privileges(username, database, minimal_privileges)
    def revoke_all_except_basic(self, username, database):
        """撤销除基本权限外的所有权限"""
        # 保留基本权限，撤销高风险权限
        privileges_to_revoke = ['DROP', 'ALTER', 'CREATE', 'INDEX', 'LOCK TABLES']
        return self.revoke_privileges(username, database, privileges_to_revoke)

注意事项

1. 安全策略：

   • 使用最小权限原则
   • 定期审计权限
   • 使用SSL/TLS加密连接
   • 密码加密存储

2. 错误处理：

   • 实现完善的异常处理
   • 记录操作日志
   • 回滚机制

3. 性能考虑：

   • 批量操作时使用事务
   • 避免频繁FLUSH PRIVILEGES
   • 合理设置连接池

这些代码提供了完整的Python操作数据库权限的解决方案,你可以根据实际需求选择使用。