3. IAM 策略文档解析与权限动作提取

为什么策略解析是权限审计的起点

IAM 策略文档是 JSON 结构，一条策略通常包含多个 Statement，每个 Statement 由 Effect、Action、Resource、Condition 四个核心字段构成。审计的目标不是“读一遍策略”，而是把嵌套结构展开成一张扁平的权限动作清单，让每一条 Allow/Deny 都能被逐行核对。下面这张图展示了解析过程中 JSON 结构到审计清单的映射关系：

解析实现：从 JSON 到权限清单

下面用 Python 实现一个策略解析器，核心逻辑是遍历 PolicyDocument 中的 Statement 数组，将每个 Statement 的 Action 展开为独立记录，并提取 Effect、Resource、Condition 字段。代码不依赖外部库，可直接运行。

import json
import csv
from io import StringIO

def parse_policy(policy_json: dict, policy_name: str = "unknown") -> list[dict]:
    """
    解析单条 IAM 策略文档，返回扁平化的权限动作清单。
    每条记录包含：policy_name, statement_index, effect, action, resource, condition
    """
    records = []
    statements = policy_json.get("Statement", [])
    if isinstance(statements, dict):
        statements = [statements]

    for idx, stmt in enumerate(statements):
        effect = stmt.get("Effect", "Unknown")
        actions = stmt.get("Action", [])
        if isinstance(actions, str):
            actions = [actions]
        resources = stmt.get("Resource", [])
        if isinstance(resources, str):
            resources = [resources]
        condition = stmt.get("Condition", None)

        for action in actions:
            for resource in resources:
                records.append({
                    "policy_name": policy_name,
                    "statement_index": idx,
                    "effect": effect,
                    "action": action,
                    "resource": resource,
                    "condition": json.dumps(condition, ensure_ascii=False) if condition else ""
                })
    return records

def flag_risks(records: list[dict]) -> list[dict]:
    """标记高风险项：通配符 Action、通配符 Resource、无 Condition 的 Allow"""
    for r in records:
        risks = []
        if r["action"].endswith(":*") or r["action"] == "*":
            risks.append("通配符Action")
        if r["resource"] == "*":
            risks.append("通配符Resource")
        if r["effect"] == "Allow" and not r["condition"]:
            risks.append("无Condition约束")
        r["risk_flags"] = "|".join(risks) if risks else ""
    return records

def export_csv(records: list[dict]) -> str:
    """导出为 CSV 字符串"""
    if not records:
        return ""
    buf = StringIO()
    writer = csv.DictWriter(buf, fieldnames=records[0].keys())
    writer.writeheader()
    writer.writerows(records)
    return buf.getvalue()

# ===== 教学示例：模拟一条 IAM 策略 =====
sample_policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": ["s3:GetObject", "s3:ListBucket"],
            "Resource": "arn:aws:s3:::data-*",
            "Condition": {"IpAddress": {"aws:SourceIp": "10.0.0.0/8"}}
        },
        {
            "Effect": "Deny",
            "Action": ["iam:DeleteUser", "iam:CreateAccessKey"],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": "ec2:*",
            "Resource": "*",
            "Condition": {"Bool": {"aws:MultiFactorAuthPresent": "true"}}
        }
    ]
}

# 执行解析
records = parse_policy(sample_policy, policy_name="sample-policy")
records = flag_risks(records)

# 输出结果
print(f"共解析出 {len(records)} 条权限动作记录\n")
for r in records:
    flag = f"  ⚠ {r['risk_flags']}" if r["risk_flags"] else ""
    print(f"[{r['effect']:5s}] {r['action']:25s} -> {r['resource']:30s}{flag}")

print("\n--- CSV 输出 ---")
print(export_csv(records))

运行上述代码，预期输出如下（教学示例结果，非真实环境实测）：

共解析出 5 条权限动作记录

[Allow] s3:GetObject              -> arn:aws:s3:::data-*              
[Allow] s3:ListBucket             -> arn:aws:s3:::data-*              
[Deny ] iam:DeleteUser            -> *                                
[Deny ] iam:CreateAccessKey       -> *                                
[Allow] ec2:*                     -> *                                ⚠ 通配符Action|通配符Resource

--- CSV 输出 ---
policy_name,statement_index,effect,action,resource,condition,risk_flags
sample-policy,0,Allow,s3:GetObject,arn:aws:s3:::data-*,{"IpAddress": {"aws:SourceIp": "10.0.0.0/8"}},
sample-policy,0,Allow,s3:ListBucket,arn:aws:s3:::data-*,{"IpAddress": {"aws:SourceIp": "10.0.0.0/8"}},
sample-policy,1,Deny,iam:DeleteUser,*,, 
sample-policy,1,Deny,iam:CreateAccessKey,*,, 
sample-policy,2,Allow,ec2:*,*,{"Bool": {"aws:MultiFactorAuthPresent": "true"}},通配符Action|通配符Resource

结果判读与审计动作

解析完成后，需要根据清单中的风险标记决定后续审计动作。以下判断逻辑可直接用于自动化审计脚本：