当前位置：首页 > news >正文

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

news 来源：原创 2025/5/2 5:39:00

import snowflake.connector
import pandas as pddef get_snowflake_permissions():# 连接Snowflake（需要替换实际凭证）conn = snowflake.connector.connect(user='<USER>',password='<PASSWORD>',account='<ACCOUNT>',warehouse='<WAREHOUSE>',role='SECURITYADMIN')# 结果容器results = {'role_grants': [],'table_privileges': [],'views': [],'masking_policies': [],'row_policies': []}# 1. 获取角色继承关系cur = conn.cursor()cur.execute("""SELECT granted_to_role, role_granted FROM SNOWFLAKE.ACCOUNT_USAGE.GRANTS_TO_ROLES WHERE PRIVILEGE = 'USAGE' AND GRANTED_ON = 'ROLE'""")results['role_grants'] = cur.fetchall()# 2. 获取表权限cur.execute("""SELECT grantee, table_catalog, table_schema, table_name, privilege_type FROM INFORMATION_SCHEMA.TABLE_PRIVILEGES""")results['table_privileges'] = cur.fetchall()# 3. 获取视图定义cur.execute("""SELECT table_catalog, table_schema, table_name, view_definition FROM INFORMATION_SCHEMA.VIEWS""")results['views'] = cur.fetchall()# 4. 获取数据掩码策略cur.execute("""SELECT policy_name, policy_body, column FROM TABLE(INFORMATION_SCHEMA.POLICY_REFERENCES())WHERE POLICY_KIND = 'MASKING_POLICY'""")results['masking_policies'] = cur.fetchall()# 5. 获取行访问策略cur.execute("""SELECT policy_name, policy_body, ref_column_name, ref_table_name FROM TABLE(INFORMATION_SCHEMA.POLICY_REFERENCES())WHERE POLICY_KIND = 'ROW_ACCESS_POLICY'""")results['row_policies'] = cur.fetchall()conn.close()return resultsdef generate_descriptions(data):reports = []# 角色继承描述for grant in data['role_grants']:reports.append(f"角色 {grant[1]} 被授予给 {grant[0]}，实现权限继承")# 表权限描述for priv in data['table_privileges']:reports.append(f"角色 {priv[0]} 在表 {priv[1]}.{priv[2]}.{priv[3]} 拥有 {priv[4]} 权限")# 视图描述for view in data['views']:reports.append(f"存在限制访问视图 {view[0]}.{view[1]}.{view[2]}，定义：{view[3]}")# 数据掩码描述for policy in data['masking_policies']:reports.append(f"列 {policy[2]} 应用数据掩码策略 {policy[0]}，策略逻辑：{policy[1]}")# 行策略描述for policy in data['row_policies']:reports.append(f"表 {policy[3]} 的 {policy[2]} 列应用行访问策略 {policy[0]}，策略逻辑：{policy[1]}")return reportsdef generate_sql_statements(data):sqls = []# 生成角色继承SQLfor grant in data['role_grants']:sqls.append(f"GRANT ROLE {grant[1]} TO ROLE {grant[0]};")# 生成表权限SQLfor priv in data['table_privileges']:sqls.append(f"GRANT {priv[4]} ON {priv[1]}.{priv[2]}.{priv[3]} TO ROLE {priv[0]};")# 生成数据掩码SQLfor policy in data['masking_policies']:sqls.extend([f"CREATE MASKING POLICY {policy[0]} AS {policy[1]};",f"ALTER TABLE {policy[2].split('.')[:3]} MODIFY COLUMN {policy[2].split('.')[3]} SET MASKING POLICY {policy[0]};"])# 生成行策略SQLfor policy in data['row_policies']:sqls.extend([f"CREATE ROW ACCESS POLICY {policy[0]} AS {policy[1]};",f"ALTER TABLE {policy[3]} ADD ROW ACCESS POLICY {policy[0]} ON ({policy[2]});"])return sqlsif __name__ == "__main__":permission_data = get_snowflake_permissions()print("=== 权限配置描述 ===")for desc in generate_descriptions(permission_data):print(desc)print("\n=== 权限重建SQL ===")for sql in generate_sql_statements(permission_data):print(sql)# 可选：将结果保存为DataFramedf_role_grants = pd.DataFrame(permission_data['role_grants'], columns=['被授权角色', '授权角色'])df_table_priv = pd.DataFrame(permission_data['table_privileges'],columns=['角色', '数据库', '模式', '表', '权限'])

输出示例：

=== 权限配置描述 ===
角色 priv_hr_rw 被授予给 dept_hr，实现权限继承
角色 priv_finance_ro 在表 finance_db.salary.reports 拥有 SELECT 权限
存在限制访问视图 finance_db.salary.v_restricted_salary，定义：SELECT employee_id, department, base_salary...
列 hr_db.employee.contacts.phone 应用数据掩码策略 phone_mask，策略逻辑：CASE WHEN CURRENT_ROLE() IN ('DEPT_SALES') THEN '***-***-' || RIGHT(val, 4)...=== 权限重建SQL ===
GRANT ROLE priv_hr_rw TO ROLE dept_hr;
GRANT SELECT ON finance_db.salary.reports TO ROLE priv_finance_ro;
CREATE MASKING POLICY phone_mask AS (val STRING)...
ALTER TABLE hr_db.employee.contacts MODIFY COLUMN phone SET MASKING POLICY phone_mask;

关键实现逻辑说明：

数据采集：通过Snowflake系统视图获取五类关键信息
- 角色继承关系
- 表级权限分配
- 视图定义及访问控制
- 动态数据掩码策略
- 行级访问策略
自然语言转换：将原始数据转换为易于理解的描述
- 使用GRANT ROLE语句解析角色继承
- 通过视图定义识别列级访问控制
- 解析策略定义描述安全逻辑
SQL重建：生成可重复执行的权限配置语句
- 保持原始权限配置的精确重建
- 处理策略定义中的Lambda表达式
- 自动生成ALTER语句应用策略