使用Python和Pandas实现的Amazon Redshift权限检查与SQL生成用于IT审计

import pandas as pd
import psycopg2
from psycopg2 import sql# 连接Redshift
conn = psycopg2.connect(host='your-cluster.endpoint.redshift.amazonaws.com',port=5439,dbname='dev',user='admin',password='your-password'
)# 权限检查函数
def check_redshift_permissions(conn):"""获取所有权限信息"""permissions = {'table_level': [],'column_level': [],'row_level': [],'data_masking': []}with conn.cursor() as cur:# 获取表级权限cur.execute("""SELECT grantee, table_schema, table_name, privilege_type FROM information_schema.table_privilegesWHERE grantee != 'PUBLIC'""")permissions['table_level'] = cur.fetchall()# 获取列级权限cur.execute("""SELECT grantee, table_schema, table_name, column_name, privilege_type FROM information_schema.column_privileges""")permissions['column_level'] = cur.fetchall()# 获取行级权限（基于视图定义）cur.execute("""SELECT viewname, definition FROM pg_views WHERE schemaname NOT IN ('pg_catalog', 'information_schema')""")for view in cur.fetchall():if ' WHERE ' in view[1].upper():permissions['row_level'].append((view[0],view[1].split('WHERE')[1].strip()))# 获取数据掩码函数cur.execute("""SELECT proname, prosrc FROM pg_proc WHERE proname LIKE 'mask%' OR proname LIKE 'dynamic_mask%'""")permissions['data_masking'] = cur.fetchall()return permissions# 获取权限数据
permissions = check_redshift_permissions(conn)# 构建自然语言描述
permission_desc = []# 表级权限处理
table_df = pd.DataFrame(permissions['table_level'], columns=['角色', 'schema', '表名', '权限类型'])
for _, row in table_df.iterrows():desc = f"角色 {row['角色']} 在表 {row['schema']}.{row['表名']} 上拥有 {row['权限类型']} 权限"sql_stmt = f"GRANT {row['权限类型']} ON {row['schema']}.{row['表名']} TO {row['角色']};"permission_desc.append(('表级权限', desc, sql_stmt))# 列级权限处理
column_df = pd.DataFrame(permissions['column_level'],columns=['角色', 'schema', '表名', '列名', '权限类型'])
for _, row in column_df.iterrows():desc = f"角色 {row['角色']} 在表 {row['schema']}.{row['表名']} 的 {row['列名']} 列上拥有 {row['权限类型']} 权限"sql_stmt = f"GRANT {row['权限类型']}({row['列名']}) ON {row['schema']}.{row['表名']} TO {row['角色']};"permission_desc.append(('列级权限', desc, sql_stmt))# 行级权限处理
for view, condition in permissions['row_level']:desc = f"视图 {view} 实施了行级过滤，条件: {condition.split('/*')[0].strip()}"sql_stmt = f"CREATE VIEW {view} AS SELECT ... WHERE {condition};"  # 需要根据实际视图定义补充permission_desc.append(('行级权限', desc, sql_stmt))# 数据掩码处理
for func_name, func_def in permissions['data_masking']:desc = f"数据掩码函数 {func_name} 实现规则: {func_def[:100]}..."sql_stmt = f"CREATE FUNCTION {func_name} ... \n{func_def};"permission_desc.append(('数据掩码', desc, sql_stmt))# 打印结果
df = pd.DataFrame(permission_desc, columns=['权限类型', '描述', 'SQL示例'])
print("权限描述：")
print(df[['权限类型', '描述']].to_markdown(index=False))print("\n对应SQL语句示例：")
print(df[['权限类型', 'SQL示例']].to_markdown(index=False))conn.close()

输出结果说明：

权限描述示例：

对应SQL语句示例：

完整权限控制SQL生成模板：

-- 角色体系
CREATE ROLE sales_dept;
CREATE ROLE hr_dept;-- 表级权限
GRANT SELECT ON TABLE sales_data.* TO sales_dept;
GRANT USAGE ON SCHEMA hr TO hr_dept;-- 列级权限
CREATE VIEW hr_limited_view AS
SELECT employee_id, name, department, CASE WHEN CURRENT_USER = 'hr_director' THEN salary ELSE NULL END AS salary
FROM employees;GRANT SELECT ON hr_limited_view TO hr_dept;-- 行级权限
CREATE VIEW sales_region_view AS
SELECT * FROM orders
WHERE region = CURRENT_SCHEMA();-- 动态数据掩码
CREATE FUNCTION mask_email(email varchar) RETURNS varchar AS $$
BEGINRETURN regexp_replace(email, '(.)(.*)@', '\1***@');
END;
$$ LANGUAGE plpgsql;CREATE VIEW customer_masked_view AS
SELECT customer_id,mask_email(email) AS email,LEFT(phone, 3) || '****' AS phone
FROM customers;-- 权限组合
GRANT sales_dept TO user1;
GRANT hr_dept TO user2;

各权限类型说明：

1. 表级权限：

   • 控制对整张表的访问
   • 典型操作：SELECT/INSERT/UPDATE/DELETE
   • 最佳实践：通过角色授权，避免直接授予用户

2. 列级权限：

   • 通过创建受限视图实现
   • 使用CASE语句动态控制列可见性
   • 配合列加密保护敏感数据

3. 行级权限：

   • 使用视图WHERE子句过滤数据
   • 动态条件：CURRENT_USER/SESSION变量
   • 可结合安全策略（Security Policy）

4. 数据掩码：

   • 使用自定义函数实现动态脱敏
   • 支持条件掩码（根据用户角色不同显示不同数据）
   • 常用方法：正则替换、数值模糊、部分隐藏

权限管理建议：

1. 使用三层角色体系：

   -- 组织级角色
   CREATE ROLE org_analyst;
   -- 部门角色
   CREATE ROLE dept_finance;
   -- 功能角色
   CREATE ROLE sensitive_data_access;GRANT sensitive_data_access TO dept_finance;
   GRANT dept_finance TO org_analyst;
   

2. 定期审计脚本：

   -- 检查权限分布
   SELECT * FROM svv_user_grants;
   -- 查看列权限
   SELECT * FROM svv_column_privileges;
   -- 审计数据访问
   SELECT * FROM svl_userlog;
   

3. 自动化清理：

   # 自动撤销过期权限示例
   def revoke_expired_permissions():expired_users = query_db("SELECT user_name FROM expired_users")for user in expired_users:execute_sql(f"REVOKE ALL PRIVILEGES ON ALL TABLES FROM {user}")