零信任架构从理论到实践：构建永不信任、始终验证的安全体系

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# 零信任架构原则

class ZeroTrustPrinciples:
    """零信任核心原则"""

    principles = {
        "Verify Explicitly": "始终验证，永不信任",
        "Use Least Privilege": "最小权限访问",
        "Assume Breach": "假设已被攻破"
    }

    def __init__(self):
        # 信任评估模型
        self.trust_score = 0
        self.context_factors = [
            'identity',
            'device_health',
            'location',
            'behavior_pattern',
            'time'
        ]

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

# 现代身份认证系统

from datetime import datetime, timedelta
import secrets

class IdentityProvider:
    """身份提供商"""

    def __init__(self):
        self.sessions = {}
        self.mfa_providers = {}

    async def authenticate(self, credentials, context):
        """多因素认证"""

        # 1. 验证身份
        user = await self._verify_identity(credentials)

        # 2. 检查设备健康
        device_trust = await self._check_device_health(context['device_id'])

        # 3. 评估上下文风险
        risk_score = await self._assess_risk(context)

        # 4. 决定认证要求
        if risk_score > 0.7:
            # 高风险：需要MFA
            mfa_required = True
        else:
            mfa_required = False

        # 5. 执行认证
        if mfa_required:
            mfa_result = await self._perform_mfa(user['id'])
            if not mfa_result['success']:
                raise AuthenticationException("MFA failed")

        # 6. 颁发令牌
        token = await self._issue_token(user, context)

        return {
            'access_token': token,
            'expires_in': 3600,
            'refresh_enabled': True
        }

    async def _issue_token(self, user, context):
        """颁发令牌"""

        # JWT令牌
        payload = {
            'user_id': user['id'],
            'roles': user['roles'],
            'permissions': user['permissions'],
            'device_id': context['device_id'],
            'location': context['location'],
            'iat': datetime.now().timestamp(),
            'exp': (datetime.now() + timedelta(hours=1)).timestamp()
        }

        # 添加信任评分
        trust_score = await self._calculate_trust_score(user, context)
        payload['trust_score'] = trust_score

        return self._encode_jwt(payload)

# 细粒度访问控制
class AccessControlEngine:
    """访问控制引擎"""

    def __init__(self, policy_engine):
        self.policy_engine = policy_engine

    async def check_access(self, request):
        """检查访问权限"""

        # 策略评估
        decision = await self.policy_engine.evaluate({
            'subject': request.subject,
            'action': request.action,
            'resource': request.resource,
            'context': request.context
        })

        return decision['allow'], decision['reason']

# 策略定义示例
ABAC_POLICIES = [
    {
        "name": "Document Access Policy",
        "conditions": {
            "resource.type": "document",
            "resource.classification": ["public", "internal", "confidential"],
            "subject.roles": ["employee", "contractor", "partner"],
            "context.location": ["office", "remote"],
            "context.time": ["business_hours", "any"]
        },
        "rules": [
            {
                "effect": "allow",
                "conditions": {
                    "resource.classification": "public",
                    "subject.roles": ["employee", "contractor", "partner"]
                }
            },
            {
                "effect": "allow",
                "conditions": {
                    "resource.classification": "internal",
                    "subject.roles": ["employee"],
                    "context.location": ["office"]
                }
            },
            {
                "effect": "deny",
                "conditions": {
                    "resource.classification": "confidential",
                    "subject.roles": ["contractor", "partner"]
                }
            }
        ]
    }
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

# 网络微分段实现

class MicroSegmentation:
    """网络微分段"""

    def __init__(self):
        self.segments = {}
        self.policies = {}

    def create_segment(self, segment_config):
        """创建网段"""

        segment_id = segment_config['id']
        self.segments[segment_id] = {
            'name': segment_config['name'],
            'workloads': [],
            'allowed_traffic': segment_config['allowed_traffic'],
            'inspection_level': segment_config['inspection_level']
        }

    def apply_policy(self, policy):
        """应用策略"""

        policy_id = policy['id']

        self.policies[policy_id] = {
            'source': policy['source_segment'],
            'destination': policy['destination_segment'],
            'ports': policy['ports'],
            'protocols': policy['protocols'],
            'action': policy['action'],  # allow/deny/inspect
            'inspection': policy.get('inspection', 'none')
        }

    def enforce_policy(self, traffic):
        """强制执行策略"""

        # 查找匹配的策略
        for policy_id, policy in self.policies.items():
            if self._traffic_matches_policy(traffic, policy):
                if policy['action'] == 'deny':
                    return False, "Policy denied"
                elif policy['action'] == 'inspect':
                    # 深度包检测
                    if not self._deep_inspect(traffic, policy['inspection']):
                        return False, "Inspection failed"

        return True, "Allowed"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

# 零信任实施阶段

class ZeroTrustRoadmap:
    """零信任实施路线图"""

    phases = [
        {
            "phase": 1,
            "name": "发现与评估",
            "duration": "1-3个月",
            "tasks": [
                "资产发现与分类",
                "数据流分析",
                "现有安全评估",
                "基线建立"
            ]
        },
        {
            "phase": 2,
            "name": "身份现代化",
            "duration": "3-6个月",
            "tasks": [
                "部署MFA",
                "实施SSO",
                "建立特权访问管理",
                "部署IAM系统"
            ]
        },
        {
            "phase": 3,
            "name": "设备信任",
            "duration": "3-6个月",
            "tasks": [
                "部署MDM",
                "实施设备健康检查",
                "建立设备信任评分",
                "部署EDR"
            ]
        },
        {
            "phase": 4,
            "name": "网络分段",
            "duration": "6-12个月",
            "tasks": [
                "实施软件定义边界",
                "部署微隔离",
                "建立零信任网络访问",
                "实施东西向流量监控"
            ]
        },
        {
            "phase": 5,
            "name": "数据保护",
            "duration": "6-12个月",
            "tasks": [
                "实施数据分类",
                "部署加密",
                "建立DLP",
                "实施CASB"
            ]
        }
    ]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

# 零信任监控系统

class ZeroTrustMonitor:
    """零信任监控"""

    def __init__(self):
        self.events = []
        self.alerts = []

    def log_access_event(self, event):
        """记录访问事件"""

        event_data = {
            'timestamp': datetime.now(),
            'subject': event['subject'],
            'action': event['action'],
            'resource': event['resource'],
            'result': event['result'],
            'context': event['context'],
            'trust_score': event['trust_score']
        }

        self.events.append(event_data)

        # 实时风险评估
        self._assess_risk(event_data)

    def _assess_risk(self, event):
        """评估风险"""

        # 异常检测
        anomalies = self._detect_anomalies(event)

        if anomalies:
            alert = {
                'level': 'high',
                'type': 'anomaly_detected',
                'details': anomalies,
                'event': event
            }

            self.alerts.append(alert)
            self._trigger_alert(alert)

    def _detect_anomalies(self, event):
        """检测异常"""

        anomalies = []

        # 1. 异常时间访问
        hour = event['timestamp'].hour
        if hour < 6 or hour > 22:
            anomalies.append("非工作时间访问")

        # 2. 异常位置访问
        if event['context']['location'] == 'unknown':
            anomalies.append("未知位置访问")

        # 3. 异常设备
        if event['context']['device_trust'] < 0.5:
            anomalies.append("低信任设备访问")

        # 4. 异常行为模式
        if self._is_unusual_behavior(event):
            anomalies.append("异常行为模式")

        return anomalies