使用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()
输出结果说明：
权限描述示例：
| 权限类型 | 描述 |
|---|---|
| 表级权限 | 角色 sales_dept 在表 sales_data.orders 上拥有 SELECT 权限 |
| 列级权限 | 角色 hr_dept 在表 hr.employees 的 salary 列上拥有 SELECT 权限 |
| 行级权限 | 视图 sales_apac_view 实施了行级过滤，条件: region = ‘APAC’ |
| 数据掩码 | 数据掩码函数 mask_ssn 实现规则: CASE WHEN … |
对应SQL语句示例：
| 权限类型 | SQL示例 |
|---|---|
| 表级权限 | GRANT SELECT ON sales_data.orders TO sales_dept; |
| 列级权限 | GRANT SELECT(salary) ON hr.employees TO hr_dept; |
| 行级权限 | CREATE VIEW sales_apac_view AS SELECT … WHERE region = ‘APAC’; |
| 数据掩码 | CREATE FUNCTION mask_ssn … |
完整权限控制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;
各权限类型说明：
-
表级权限：
- 控制对整张表的访问
- 典型操作：SELECT/INSERT/UPDATE/DELETE
- 最佳实践：通过角色授权，避免直接授予用户
-
列级权限：
- 通过创建受限视图实现
- 使用CASE语句动态控制列可见性
- 配合列加密保护敏感数据
-
行级权限：
- 使用视图WHERE子句过滤数据
- 动态条件：CURRENT_USER/SESSION变量
- 可结合安全策略（Security Policy）
-
数据掩码：
- 使用自定义函数实现动态脱敏
- 支持条件掩码（根据用户角色不同显示不同数据）
- 常用方法：正则替换、数值模糊、部分隐藏
权限管理建议：
-
使用三层角色体系：
-- 组织级角色 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; -
定期审计脚本：
-- 检查权限分布 SELECT * FROM svv_user_grants; -- 查看列权限 SELECT * FROM svv_column_privileges; -- 审计数据访问 SELECT * FROM svl_userlog; -
自动化清理：
# 自动撤销过期权限示例 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}")