Network Firewall 部署小指南 (六) 利用 Amazon Bedrock AI 实现Network Firewall规则冲突的实时检测与智能分析

用户编辑 Rule Group → Save
    │
    ▼
CloudTrail 记录 API → EventBridge 匹配 → 触发 Lambda
    │
    ▼
┌─────────────────────────────────────────────┐
│  Lambda 内部执行步骤                          │
│                                             │
│  Step 1: 获取所有 Rule Group 规则（API 调用）  ｜
│  Step 2: 查询哪些 Rule Group 已在 Policy 中   ｜
│  Step 3: 代码做冲突检测（CIDR/端口/域名计算）    |
│  Step 4: 生成 SVG 可视化图 → 存 S3            |
│  Step 5: 调用 Bedrock AI 分析                |
│  Step 6: 有冲突 → SNS 发邮件                  |
└─────────────────────────────────────────────┘

# --- RFC1918 私有地址定义（用于判断 $EXTERNAL_NET 变量语义）---
RFC1918_NETWORKS = [
    ipaddress.ip_network('10.0.0.0/8'),
    ipaddress.ip_network('172.16.0.0/12'),
    ipaddress.ip_network('192.168.0.0/16'),
]


def is_rfc1918(cidr_str):
    """判断一个 CIDR 是否属于 RFC1918 私有地址"""
    cidr_str = cidr_str.strip('[]')
    if '/' not in cidr_str and re.match(r'^\d+\.\d+\.\d+\.\d+$', cidr_str):
        cidr_str = cidr_str + '/32'
    try:
        net = ipaddress.ip_network(cidr_str, strict=False)
        return any(net.subnet_of(rfc) for rfc in RFC1918_NETWORKS)
    except ValueError:
        return False


def cidr_overlaps(cidr1, cidr2):
    """
    检查两个 CIDR 是否有重叠，并理解 Suricata 变量的语义。

    返回值：
    - 'exact'    : 完全相同
    - 'subset'   : cidr1 是 cidr2 的子集
    - 'superset' : cidr1 是 cidr2 的超集
    - 'partial'  : 部分重叠
    - 'none'     : 不重叠
    - 'unknown'  : 无法确定（如 @变量引用）
    - 'possible' : 可能重叠但无法确认

    关键创新点：
    - $EXTERNAL_NET vs RFC1918 私有地址 → 判定为不重叠（公网≠内网）
    - $HOME_NET vs 公网地址 → 判定为不重叠
    - @IPSet 变量引用 → 无法确定范围，返回 unknown 不报冲突
    - 支持逗号分隔的 IP 列表（如 [1.1.1.1,2.2.2.2]）
    """
    # 统一大小写
    cidr1_lower = cidr1.lower()
    cidr2_lower = cidr2.lower()

    # 两个都是 any → 完全相同
    if cidr1_lower == 'any' and cidr2_lower == 'any':
        return 'exact'

    # 一个是 any → 超集关系
    if cidr1_lower == 'any' or cidr2_lower == 'any':
        return 'superset'

    # ---- $EXTERNAL_NET 的语义处理 ----
    # $EXTERNAL_NET 通常定义为"非 RFC1918"，即公网地址
    # 所以 $EXTERNAL_NET vs 10.x.x.x → 不重叠
    if cidr1 == '$EXTERNAL_NET' or cidr2 == '$EXTERNAL_NET':
        other = cidr2 if cidr1 == '$EXTERNAL_NET' else cidr1
        if is_rfc1918(other):
            return 'none'  # 公网 vs 私网 → 不重叠
        if other == '$EXTERNAL_NET':
            return 'exact'
        if other == '$HOME_NET':
            return 'none'  # HOME_NET 通常是私网
        return 'possible'

    # ---- $HOME_NET 的语义处理 ----
    if cidr1 == '$HOME_NET' or cidr2 == '$HOME_NET':
        other = cidr2 if cidr1 == '$HOME_NET' else cidr1
        if other == '$HOME_NET':
            return 'exact'
        if other == '$EXTERNAL_NET':
            return 'none'
        if is_rfc1918(other):
            return 'possible'  # 都是私网，可能重叠
        return 'none'  # 另一方是公网，不重叠

    # ---- @IPSet 变量引用（如 @PreproprivateEC2AZ1）----
    # 无法确定实际 IP 范围，返回 unknown 避免误报
    if cidr1.startswith('@') or cidr2.startswith('@'):
        if cidr1 == cidr2:
            return 'exact'
        return 'unknown'

    # ---- 逗号分隔的 IP 列表或方括号包裹 ----
    if ',' in cidr1 or ',' in cidr2 or cidr1.startswith('[') or cidr2.startswith('['):
        return check_ip_list_overlap(cidr1, cidr2)

    # ---- 补全没有子网掩码的单 IP ----
    if '/' not in cidr1 and re.match(r'^\d+\.\d+\.\d+\.\d+$', cidr1):
        cidr1 = cidr1 + '/32'
    if '/' not in cidr2 and re.match(r'^\d+\.\d+\.\d+\.\d+$', cidr2):
        cidr2 = cidr2 + '/32'

    # ---- 标准 CIDR 子网计算 ----
    try:
        net1 = ipaddress.ip_network(cidr1, strict=False)
        net2 = ipaddress.ip_network(cidr2, strict=False)
    except ValueError:
        return 'unknown'

    if net1 == net2:
        return 'exact'
    elif net1.subnet_of(net2):
        return 'subset'
    elif net2.subnet_of(net1):
        return 'superset'
    elif net1.overlaps(net2):
        return 'partial'
    else:
        return 'none'


def check_ip_list_overlap(cidr1, cidr2):
    """检查逗号分隔的 IP 列表是否有重叠"""
    def parse_ip_list(s):
        s = s.strip('[]')
        parts = [p.strip() for p in s.split(',')]
        networks = []
        for p in parts:
            try:
                if '/' not in p:
                    p = p + '/32'
                networks.append(ipaddress.ip_network(p, strict=False))
            except ValueError:
                pass
        return networks

    nets1 = parse_ip_list(cidr1) if ',' in cidr1 else [ipaddress.ip_network(cidr1.strip('[]') + ('/32' if '/' not in cidr1 else ''), strict=False)]
    nets2 = parse_ip_list(cidr2) if ',' in cidr2 else [ipaddress.ip_network(cidr2.strip('[]') + ('/32' if '/' not in cidr2 else ''), strict=False)]

    for n1 in nets1:
        for n2 in nets2:
            if n1.overlaps(n2):
                return 'exact' if n1 == n2 else 'partial'
    return 'none'


def deep_match_overlaps(dm1, dm2):
    """
    检查两条规则的深度匹配条件（Suricata options）是否有重叠。

    关键逻辑：
    - 如果两条规则匹配不同的域名（如 .google.com vs .baidu.com）→ 不冲突
    - 如果匹配不同的 content_type（如 text/html vs audio/mpeg）→ 不冲突
    - 如果匹配不同的 HTTP 方法（如 GET vs POST）→ 不冲突

    这是消除误报的关键环节：即使 IP/端口层面有重叠，
    但深度条件不同意味着匹配的是完全不同类型的流量。
    """
    if not dm1 and not dm2:
        return True

    # http.host 域名对比
    if dm1.get('http_host') and dm2.get('http_host'):
        if not domain_overlaps(dm1['http_host'], dm2['http_host']):
            return False  # 不同域名 → 不同流量 → 不冲突

    # http.content_type 对比
    if dm1.get('http_content_type') and dm2.get('http_content_type'):
        if dm1['http_content_type'] != dm2['http_content_type']:
            return False

    # http.method 对比
    if dm1.get('http_method') and dm2.get('http_method'):
        if dm1['http_method'] != dm2['http_method']:
            return False

    return True


def domain_overlaps(domain1, domain2):
    """检查两个域名匹配模式是否有重叠（考虑子域名关系）"""
    if domain1 == domain2:
        return True
    d1 = domain1.lstrip('.')
    d2 = domain2.lstrip('.')
    if d1.endswith('.' + d2) or d2.endswith('.' + d1):
        return True
    if d1 == d2:
        return True
    return False

AI_PROMPT_TEMPLATE = """You are an AWS Network Firewall security expert. Analyze the following rule conflicts detected in a Network Firewall policy deployment in eu-west-1.
## Important Context - Rule Evaluation Order (STRICT_ORDER mode):
- Rules are evaluated by Rule Group priority (lower number = higher priority).
- Rule Groups already in the Policy have HIGHER priority than newly added ones.
- Once a packet matches a rule, evaluation STOPS — subsequent lower-priority rules are NOT checked for that packet.
- Key implication: If Rule Group A (higher priority) passes traffic from 10.1.1.0/24, and Rule Group B (lower priority) drops traffic from 10.1.0.0/16, the actual behavior is:
  * 10.1.1.0/24 traffic → matches Rule A first → PASS (Rule B never sees it)
  * 10.1.2.0/24, 10.1.3.0/24, etc. → does NOT match Rule A → falls through to Rule B → DROP
  * This forms a "deny-all with exception" pattern — BOTH rules are effective, just for different subsets of traffic.
## Conflict Type Definitions:
- CIDR_SUBSET_OVERLAP: Two rules have CIDR subset/superset relationship with different actions. This is a CIDR subset overlap that may indicate intentional layered policy design for different application/business needs.
- ACTION_CONFLICT: Two rules match exactly the same traffic (same CIDR, same port, same protocol) but have opposite actions.
- DOMAIN_ACTION_CONFLICT: Domain list rules with contradicting allow/deny for the same domain.
## Current Rules:
{rules_summary}
## Detected Conflicts:
{conflicts_summary}
Please provide your analysis in Chinese (中文):
1. 意图分析: 判断每个冲突是有意设计（如"大范围拒绝 + 小范围例外放行"模式）还是配置错误。精确说明每条规则实际对哪些流量生效。
2. 风险评估: 对每个冲突评估风险等级（高/中/低）。如果是有意的分层策略设计则风险较低，如果是无意的配置错误则风险较高。说明潜在影响。
3. 修复建议: 对每个冲突给出 2-3 个修复方向和思路，只描述策略调整的方向（如"调整优先级"、"缩小网段范围"、"合并到同一个 Rule Group"等），不要给出具体的 Suricata 规则语句或代码示例。
4. 总结: 一段简短的总结，适合发送给管理层。
格式要求：
- 直接使用 Rule Group 名称来引用规则，不要使用"规则 A"、"规则 B"这样的标签
- 明确说明哪个 Rule Group 在 Policy 中优先级更高（已部署的优先级高于新添加的）
- 精确描述流量匹配行为：哪些流量命中哪条规则、结果是什么
- 不要使用任何 markdown 格式符号（如反引号、星号、井号等），输出纯文本格式
请用中文回复，规则内容保持英文原文。"""