一、最小可用的零信任引擎长什么样

把"验证每次访问"定为第一性原理。工程上，它落地为一个独立的策略决策点（PDP）：接收一次访问请求（主体身份、设备状态、目标资源），结合策略库与风险评分，输出 allow / deny。执行点（PEP）可以是网关、服务网格 sidecar 或 API 网关，只管放行或拦截，不持有判断逻辑。

本篇配套zero_trust_engine.py实现了一个零依赖的 PDP：它用标签描述主体与资源，用策略规则做匹配，并叠加"设备是否合规""是否训练数据出域"等风险因子。把它接到真实系统只需把策略源换成数据库、把 verdict 回传给网关。

这里要把两个角色讲清：PEP（策略执行点）是真正拦流量的地方，可能是公司出口网关、K8s 里的 sidecar、或某个微服务的 API 网关；PDP（策略决策点）是"大脑"，只算 allow/deny 并返回依据。把判断和拦截分开，好处是同一套授权逻辑能复用在网关、服务网格、数据库代理等多种执行点，省去各处重写规则。

二、策略即代码：用标签代替 IP

零信任的关键转身是**从"基于 IP/网段"转到"基于身份标签"**。AI 团队最怕的就是"研发网 10.1.x.x 能访问模型服务"这种写法——一旦容器漂移、IP 复用，策略立刻失真。改用标签（role=researcher、env=training、data=pii）后，策略与网络位置彻底解耦，自动扩缩容也不会破坏授权。

rules = [ # 研发可访问模型服务，但禁止触碰含 PII 的财务库 Policy(subject={"role":"researcher"}, resource={"svc":"model-svc"}, allow=True), Policy(subject={"role":"researcher"}, resource={"data":"pii"}, allow=False), # 训练任务允许出域拉取公开权重，但训练样本不得出域 Policy(subject={"env":"training"}, resource={"data":"public-weights"}, allow=True), Policy(subject={"env":"training"}, resource={"data":"train-set"}, allow=False), ]

策略描述要像写给人看的规则，而不是防火墙 ACL。可读、可评审、可回滚，才谈得上"治理"。

三、网络微分段：把爆炸半径关进格子

零信任管"谁能访问"，微分段管"即使被攻陷也跑不远"。传统 VLAN 是粗粒度的大房间，一台机器失陷就能横向扫整个段。微分段给每个工作负载贴标签，按"服务到服务"的必需最小连接画格子，默认拒绝一切、仅显式放行。segment_network.py演示了如何根据标签生成段间规则并做冲突校验。

四、把代码跑起来：两个脚本做了什么

zero_trust_engine.py接收一条请求（如 researcher 访问 model-svc），依次做：标签匹配 → 命中显式 deny 则直接拒绝 → 命中 allow 且设备合规则放行 → 否则拒绝，并打印决策依据。它内置了一个非法场景——researcher 试图读 pii 财务库，引擎返回 deny 并说明原因，这正是零信任的硬约束。

segment_network.py则根据工作负载标签生成段间规则表，并对"同段内是否过度互信""是否存在悬空放行"做静态校验，输出一份可读的分段报告。两者都零依赖，python xxx.py直接跑。

# 运行示例 python zero_trust_engine.py python segment_network.py

先能手写最小 PDP 与分段器，才看得懂商业零信任产品在替你做什么；选型讨论才不会停在"厂商说好"的层面。

zero_trust_engine.py

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ zero_trust_engine.py — 最小可运行零信任策略决策点（PDP）演示 用标签描述主体与资源，按策略规则做匹配，叠加设备合规/数据出域风险因子， 输出 allow / deny 与决策依据。零依赖。运行：python zero_trust_engine.py """ from dataclasses import dataclass, field from typing import Dict, List @dataclass class Policy: subject: Dict[str, str] resource: Dict[str, str] allow: bool # 策略即代码：基于身份标签，而非 IP RULES: List[Policy] = [ Policy(subject={"role": "researcher"}, resource={"svc": "model-svc"}, allow=True), Policy(subject={"role": "researcher"}, resource={"data": "pii"}, allow=False), Policy(subject={"env": "training"}, resource={"data": "public-weights"}, allow=True), Policy(subject={"env": "training"}, resource={"data": "train-set"}, allow=False), ] def _match(rule_tags: Dict[str, str], req_tags: Dict[str, str]) -> bool: return all(req_tags.get(k) == v for k, v in rule_tags.items()) def decide(subject: Dict[str, str], resource: Dict[str, str], device_compliant: bool = True) -> Dict: """PDP 核心：返回决策与依据列表。""" reasons: List[str] = [] # 风险因子 1：设备不合规 -> 直接拒绝 if not device_compliant: reasons.append("设备未通过合规检查（终端健康度不足）") return {"verdict": "deny", "reasons": reasons} # 逐条匹配策略，显式 deny 优先 for rule in RULES: if _match(rule.subject, subject) and _match(rule.resource, resource): if rule.allow: reasons.append("命中允许规则，且设备合规") return {"verdict": "allow", "reasons": reasons} reasons.append("命中显式拒绝规则（最小权限约束）") return {"verdict": "deny", "reasons": reasons} reasons.append("无匹配允许规则，按默认拒绝") return {"verdict": "deny", "reasons": reasons} def main(): print("=" * 56) print("零信任策略决策点（PDP）演示") print("=" * 56) cases = [ ("researcher 访问 model-svc（设备合规）", {"role": "researcher"}, {"svc": "model-svc"}, True), ("researcher 读取 PII 财务库（应拒绝）", {"role": "researcher"}, {"data": "pii"}, True), ("训练任务拉取公开权重", {"env": "training"}, {"data": "public-weights"}, True), ("训练任务外传训练样本（应拒绝）", {"env": "training"}, {"data": "train-set"}, True), ("员工用未合规设备访问模型服务（应拒绝）", {"role": "researcher"}, {"svc": "model-svc"}, False), ] for desc, subj, res, ok in cases: r = decide(subj, res, ok) print(f"\n场景: {desc}") print(f" 决策: {r['verdict'].upper()}") for reason in r["reasons"]: print(f" - {reason}") print("\n" + "=" * 56) if __name__ == "__main__": main()

segment_network.py

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ segment_network.py — 网络微分段规则生成与静态校验（可运行演示） 根据工作负载标签生成段间"默认拒绝、仅放行必需连接"的规则表， 并对过度互信/悬空放行做静态校验。零依赖。运行：python segment_network.py """ from dataclasses import dataclass, field from typing import Dict, List @dataclass class Workload: name: str seg: str # 所属分段 labels: Dict[str, str] # 工作负载清单（标签驱动） WORKLOADS: List[Workload] = [ Workload("notebook-01", "research", {"role": "researcher"}), Workload("trainer-01", "training", {"env": "training"}), Workload("model-svc", "model", {"svc": "model-svc"}), Workload("finance-db", "finance", {"data": "pii"}), ] # 显式放行的必需连接（source_seg -> target_seg） ALLOW: List[tuple] = [ ("research", "model"), ("training", "model"), ] def build_rules() -> List[dict]: rules = [] segs = sorted({w.seg for w in WORKLOADS}) for a in segs: for b in segs: allowed = (a, b) in ALLOW # 同段内仅允许"协作必要"的连接，这里演示为默认同段放行（可按需收紧） if a == b: rules.append({"from": a, "to": b, "action": "allow", "note": "同段（后续可再微分）"}) else: rules.append({ "from": a, "to": b, "action": "allow" if allowed else "deny", "note": "必需最小连接" if allowed else "默认拒绝", }) return rules def validate(rules: List[dict]) -> List[str]: warns = [] segs = {r["from"] for r in rules} # 1) 每个非 finance 段是否都默认拒绝通往 finance（防 PII 横向） for s in segs: if s == "finance": continue hit = [r for r in rules if r["from"] == s and r["to"] == "finance"] for r in hit: if r["action"] != "deny": warns.append(f"风险: 段 '{s}' 可直达 finance(PII)，应默认拒绝") # 2) 悬空放行：allow 但目标段不存在 existing = segs for r in rules: if r["action"] == "allow" and r["to"] not in existing: warns.append(f"悬空放行: {r['from']} -> {r['to']} 目标段不存在") if not warns: warns.append("校验通过：无过度互信、无悬空放行。") return warns def main(): print("=" * 56) print("网络微分段规则生成与校验") print("=" * 56) rules = build_rules() print("\n段间规则表（默认拒绝，仅放行必需连接）：") for r in rules: tag = "✓" if r["action"] == "allow" else "✗" print(f" {tag} {r['from']:>9} -> {r['to']:<9} [{r['action']}] {r['note']}") print("\n静态校验结果：") for w in validate(rules): print(f" - {w}") print("=" * 56) if __name__ == "__main__": main()

五、落到真实环境的衔接点

演示脚本把判断逻辑具象化了，接生产时只需把三处换成真实组件：策略源（数据库/OPA）、身份源（SSO/设备合规平台）、执行点（服务网格或 API 网关）。核心判断算法不变，换来换去的是数据接入。这也是为什么架构篇强调"身份即边界"——它让上层逻辑与底层网络解耦，迁移成本被压到最低。

网络选型的迭代成本和"耦合面"成正比。把授权封装成独立 PDP，未来换网关只是改一处调用，而不是重构全链路。

落地节奏上，建议先在"测试环境 + 影子模式"跑 PDP：它照常算出 verdict，但先不真正拦截，只把决策与现有 ACL 结果做比对，确认零误杀后再切到强制模式。微分段也一样——先观察段间真实流量，再逐步把"默认拒绝"从低风险段推广到核心段，避免一刀切把业务打断。