AI 辅助 K8s 网络策略智能生成与安全审计：从手动配置到自动化防护

一、K8s 网络策略的配置困境：一条规则遗漏就是一次安全事件

Kubernetes 网络策略（NetworkPolicy）是集群内部流量隔离的核心机制，但它的配置极其繁琐且容易出错。一个中型集群可能有上百个 Service、数千个 Pod，服务间的调用关系错综复杂。手动编写 NetworkPolicy 需要精确指定每个命名空间的每个标签选择器，遗漏一条规则就可能导致数据库 Pod 暴露给所有命名空间。更棘手的是，业务频繁变更——新服务上线、旧服务下线、标签修改——每次变更都可能让已有的网络策略失效。

AI 辅助网络策略生成通过分析集群的实际流量模式，自动推导服务间的合法通信关系，生成最小权限的网络策略，并持续审计策略与实际流量的偏差。

二、智能网络策略生成架构

flowchart TD
    A[集群流量采集] --> A1[Service Mesh 可观测数据]
    A --> A2[NetworkPolicy 审计日志]
    A --> A3[DNS 查询日志]
    A1 --> B[流量关系图谱]
    A2 --> B
    A3 --> B
    B --> C[AI 策略推导引擎]
    C --> C1[合法流量模式识别]
    C --> C2[异常流量检测]
    C --> C3[策略缺口发现]
    C1 --> D[NetworkPolicy 生成]
    C2 --> E[安全告警]
    C3 --> D
    D --> F[策略预览与审批]
    F --> G[灰度发布与验证]

2.1 集群流量关系采集

# traffic_collector.py — 集群流量关系采集
# 设计意图：从 Service Mesh 和 DNS 日志中提取服务间通信关系

from dataclasses import dataclass
from collections import defaultdict

@dataclass
class TrafficEdge:
    src_namespace: str
    src_service: str
    dst_namespace: str
    dst_service: str
    dst_port: int
    protocol: str
    request_count: int
    last_seen: str

class TrafficCollector:
    def __init__(self):
        self.edges: list[TrafficEdge] = []
        self.service_labels: dict[str, dict] = {}  # service → labels

    def parse_istio_access_log(self, log_line: str) -> TrafficEdge | None:
        """从 Istio 访问日志提取流量关系"""
        import json
        try:
            entry = json.loads(log_line)
        except json.JSONDecodeError:
            return None

        source = entry.get("source", {})
        destination = entry.get("destination", {})

        return TrafficEdge(
            src_namespace=source.get("namespace", "unknown"),
            src_service=source.get("workload", "unknown"),
            dst_namespace=destination.get("namespace", "unknown"),
            dst_service=destination.get("workload", "unknown"),
            dst_port=int(destination.get("port", 0)),
            protocol=entry.get("protocol", "TCP"),
            request_count=1,
            last_seen=entry.get("timestamp", ""),
        )

    def build_traffic_graph(self) -> dict[str, list[TrafficEdge]]:
        """构建命名空间维度的流量关系图"""
        graph: dict[str, list[TrafficEdge]] = defaultdict(list)
        for edge in self.edges:
            key = f"{edge.src_namespace}→{edge.dst_namespace}"
            graph[key].append(edge)
        return graph

2.2 AI 策略推导引擎

# policy_generator.py — AI 网络策略推导引擎
# 设计意图：基于流量关系图，用 AI 推导最小权限网络策略

import json

POLICY_GENERATION_PROMPT = """你是一个 Kubernetes 网络安全专家。根据以下服务间流量关系，生成最小权限的 NetworkPolicy。

命名空间: {namespace}
该命名空间的服务: {services}
入站流量来源: {inbound_edges}
出站流量目标: {outbound_edges}

当前已有的 NetworkPolicy: {existing_policies}

要求:
1. 默认拒绝所有入站和出站流量
2. 仅允许上述流量关系中存在的合法通信
3. 使用标签选择器精确匹配源和目标
4. 为每个端口和协议创建独立的规则
5. 检查已有策略是否存在缺口或冗余

输出 JSON:
{{"policies": [{{"name": "...", "spec": {{...}}}}], "gaps": [{{"description": "..."}}], "redundancies": [{{"policy": "...", "reason": "..."}}]}}"""

async def generate_network_policies(
    namespace: str,
    traffic_graph: dict,
    existing_policies: list[dict],
    llm_client,
) -> dict:
    """AI 推导网络策略"""
    inbound = [e for e in traffic_graph.get("inbound", [])]
    outbound = [e for e in traffic_graph.get("outbound", [])]
    services = list(set(
        [e.dst_service for e in inbound] + [e.src_service for e in outbound]
    ))

    prompt = POLICY_GENERATION_PROMPT.format(
        namespace=namespace,
        services=json.dumps(services),
        inbound_edges=json.dumps(inbound[:20], default=str),
        outbound_edges=json.dumps(outbound[:20], default=str),
        existing_policies=json.dumps(existing_policies),
    )

    response = await llm_client.chat(prompt, temperature=0.1)
    try:
        return json.loads(response)
    except json.JSONDecodeError:
        return {"policies": [], "gaps": [], "redundancies": []}

2.3 策略安全审计

# policy_auditor.py — 网络策略安全审计
# 设计意图：检测策略与实际流量的偏差，发现安全风险

from dataclasses import dataclass

@dataclass
class AuditFinding:
    severity: str  # critical, high, medium, low
    category: str  # gap, redundancy, misconfiguration
    description: str
    namespace: str
    policy_name: str | None
    recommendation: str

class NetworkPolicyAuditor:
    def audit(
        self,
        traffic_graph: dict,
        policies: list[dict],
    ) -> list[AuditFinding]:
        """审计网络策略与实际流量的偏差"""
        findings = []

        # 检查1: 无策略的命名空间
        covered_ns = {p["metadata"]["namespace"] for p in policies}
        all_ns = set()
        for key in traffic_graph:
            for ns in key.split("→"):
                all_ns.add(ns)

        for ns in all_ns - covered_ns:
            findings.append(AuditFinding(
                severity="critical",
                category="gap",
                description=f"命名空间 {ns} 没有任何 NetworkPolicy，所有流量默认允许",
                namespace=ns,
                policy_name=None,
                recommendation=f"为 {ns} 创建默认拒绝策略",
            ))

        # 检查2: 过于宽松的规则
        for policy in policies:
            spec = policy.get("spec", {})
            for ingress in spec.get("ingress", []):
                for rule in ingress.get("from", []):
                    if not rule.get("namespaceSelector") and not rule.get("podSelector"):
                        findings.append(AuditFinding(
                            severity="high",
                            category="misconfiguration",
                            description=f"策略 {policy['metadata']['name']} 的入站规则没有选择器，允许所有流量",
                            namespace=policy["metadata"]["namespace"],
                            policy_name=policy["metadata"]["name"],
                            recommendation="添加 namespaceSelector 或 podSelector 限制来源",
                        ))

        return sorted(findings, key=lambda x: ["critical", "high", "medium", "low"].index(x.severity))

三、灰度发布与验证流程

3.1 策略干运行模式

# policy_dryrun.py — 策略干运行验证
# 设计意图：在应用策略前模拟其效果，检测是否阻断合法流量

import subprocess
import json

def dry_run_policy(
    policy: dict,
    namespace: str,
    test_traffic: list[dict],
) -> list[dict]:
    """干运行模式验证策略效果"""
    results = []

    for traffic in test_traffic:
        src_label = traffic.get("src_labels", {})
        dst_label = traffic.get("dst_labels", {})
        port = traffic.get("port", 0)

        # 模拟策略匹配逻辑
        allowed = check_policy_allows(policy, src_label, dst_label, port)

        results.append({
            "traffic": traffic,
            "allowed": allowed,
            "expected": traffic.get("expected", True),
            "match": allowed == traffic.get("expected", True),
        })

    mismatches = [r for r in results if not r["match"]]
    return mismatches


def check_policy_allows(
    policy: dict,
    src_labels: dict,
    dst_labels: dict,
    port: int,
) -> bool:
    """模拟 NetworkPolicy 匹配逻辑"""
    spec = policy.get("spec", {})

    # 检查 podSelector 是否匹配目标
    pod_selector = spec.get("podSelector", {})
    if pod_selector.get("matchLabels"):
        for k, v in pod_selector["matchLabels"].items():
            if dst_labels.get(k) != v:
                return True  # 策略不适用于此 Pod，默认允许

    # 检查入站规则
    for ingress in spec.get("ingress", []):
        for rule in ingress.get("from", []):
            ns_sel = rule.get("namespaceSelector", {})
            pod_sel = rule.get("podSelector", {})
            if matches_selector(src_labels, ns_sel, pod_sel):
                for port_rule in ingress.get("ports", []):
                    if port_rule.get("port") == port:
                        return True

    return False  # 默认拒绝

四、边界分析与架构权衡

流量采集的盲区：Service Mesh 只能采集经过 Sidecar 代理的流量，HostNetwork 模式的 Pod 和 kube-system 命名空间的流量可能被遗漏。需要补充 CNI 层的流量日志来覆盖盲区。

AI 生成策略的可信度：AI 推导的策略可能遗漏低频但合法的流量（如定时任务、管理端口）。建议先以"审计模式"运行，只报告策略缺口，不自动应用策略，人工确认后再开启"执行模式"。

策略数量膨胀：每个服务一个策略会导致策略数量爆炸，增加 API Server 的负担。建议按命名空间维度合并策略，减少策略总数。

流量基线的冷启动：新上线的服务没有历史流量数据，AI 无法推导其合法通信关系。建议为新服务先配置宽松策略，积累 7 天流量数据后再收紧。

五、总结

AI 辅助 K8s 网络策略生成将安全配置从"手动编写"升级为"流量驱动自动推导"，通过采集集群实际流量关系，用 AI 推导最小权限策略，并持续审计策略与流量的偏差。落地建议：先以审计模式运行，只报告缺口不自动应用；补充 CNI 层流量日志覆盖盲区；按命名空间合并策略减少数量；新服务先宽松后收紧，积累流量基线后再收紧策略。