SASE 与 CASB 实战：安全访问服务边界与云访问安全的落地

传统架构（回程模式）：
  远程用户 ──VPN──▶ 总部防火墙 ──▶ 数据中心应用
                       └──▶ 分支访问 SaaS 也要绕行
  问题：延迟高、带宽浪费、VPN 漏洞（CVE-2021-34506 等）、策略不随身份

SASE 架构（边缘就近）：
  远程用户 ──▶ 就近 PoP（融合安全能力）──▶ SaaS / 私有应用
                    │
              身份验证 + 策略 + 检查全部在边缘完成
  优势：低延迟、安全内嵌、身份驱动

SASE 两大支柱：
┌─────────────────────────────────────┐
│ 网络能力（Network）                  │
│  · SD-WAN（广域网优化）              │
│  · 全球边缘 PoP 就近接入             │
│  · 云骨干网传输                     │
├─────────────────────────────────────┤
│ 安全能力（Security）                 │
│  · ZTNA 零信任网络访问               │
│  · SWG 安全 Web 网关                 │
│  · CASB 云访问安全代理               │
│  · FWaaS 防火墙即服务                │
│  · DLP 数据防泄漏                    │
└─────────────────────────────────────┘
统一由云端控制台编排策略，身份 + 设备 + 应用驱动

ZTNA 每次访问都验证：
  1. 身份（Who）：用户是谁（MFA、SSO）
  2. 设备（What）：设备是否合规（补丁、磁盘加密）
  3. 上下文（Where/When）：位置、时间、风险评分
  验证通过才建立到目标应用的加密隧道（最小授权）

# Cloudflare Tunnel：把内网服务暴露为私密应用（无公网端口）
cloudflared tunnel login
cloudflared tunnel create my-app-tunnel
# 配置 config.yml：映射到内网服务
cloudflared tunnel route dns my-app-tunnel app.example.com
cloudflared tunnel run my-app-tunnel

# Tailscale（基于 WireGuard 的网格 VPN）
# 特点：全网状加密、身份驱动访问、ACL 控制
tailscale up --ssh
# ACL：只允许某用户组访问特定机器端口
# 原则：默认拒绝，显式允许

# minimal_ztna.py — 身份感知代理的最小实现
import functools
import jwt

def identity_gate(required_role: str):
    """装饰器：验证 JWT 身份 + 角色后才代理到内网应用。"""
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(request, *args, **kwargs):
            token = request.headers.get("Authorization", "").replace("Bearer ", "")
            try:
                payload = jwt.decode(token, SECRET, algorithms=["HS256"])
            except Exception:
                return {"error": "unauthorized"}, 401

            # 身份 + 设备 + 上下文三重校验
            if payload.get("role") != required_role:
                return {"error": "forbidden"}, 403
            if not device_compliant(payload.get("device_id")):
                return {"error": "device_not_compliant"}, 403
            if high_risk_context(payload):
                return {"error": "risky_context"}, 401

            return fn(request, payload)   # 放行，带上身份信息
        return wrapper
    return decorator


@identity_gate("finance.auditor")
def proxy_to_internal_api(request, identity):
    """只允许财务审计角色访问内部报表 API。"""
    return forward_to_internal("http://10.0.0.5:8080", request.path)

SaaS 应用（Salesforce/Google/微信工作台等）三大风险：
  1. 影子 IT：员工自行注册云应用，IT 不知道
  2. 权限失控：离职账号未清理、过度授权
  3. 数据外泄：敏感数据上传、DLP 缺失

CASB 作为"中间层"控制云应用访问与数据：
  用户 → CASB → SaaS
          │
    权限治理 / DLP / 行为分析 / 影子应用发现

# casb_dlp.py — 基于 SaaS API 的数据治理
def scan_uploaded_file_for_pii(file_meta, content) -> dict:
    """CASB 扫描上传至网盘的敏感数据。"""
    from detection import classify_pii
    pii = classify_pii(content)
    policy = {
        "PII_PASSport": "block",
        "PII_CreditCard": "block+alert",
        "PII_Phone": "alert_only",
    }
    actions = []
    for kind, found in pii.items():
        if found and policy.get(kind) == "block":
            actions.append(f"block:{kind}")
        if found and "alert" in policy.get(kind, ""):
            actions.append(f"alert:{kind}")
    return {"actions": actions, "detected": pii}

def enforce_saas_policy(file_meta, content, saas_api):
    """依据 DLP 策略执行：阻断或告警。"""
    result = scan_uploaded_file_for_pii(file_meta, content)
    if any(a.startswith("block") for a in result["actions"]):
        saas_api.abort_upload(file_meta["id"])
        alert_security(file_meta["user"], result["detected"])
        return "blocked"
    return "allowed"

def discover_shadow_it(access_logs, known_apps: set) -> list[dict]:
    """从代理/日志中发现未批准的云应用访问。"""
    shadow = []
    for app in extract_cloud_apps(access_logs):
        if app not in known_apps:
            shadow.append({
                "app": app,
                "users": distinct_users_for(app),
                "bandwidth": bandwidth_for(app),
                "risk": assess_app_risk(app),
            })
    return sorted(shadow, key=lambda s: s["risk"], reverse=True)

策略编排目标：
  · 一个控制台管理所有边缘能力
  · 策略以"身份/设备/应用"为对象，而非 IP
  · 一次定义，全局分发到所有 PoP

策略示例：
  if 用户.role in {finance} and 设备.合规 and 应用 == "财务系统":
      允许访问 + DLP 检查
  else if 应用 == "高风险SaaS" and 未批准:
      阻断 + 告警

# sase_policy.yml — 声明式 SASE 策略（示意）
version: "1.0"
policies:
  - name: finance-app-access
    action: allow
    when:
      identity: {roles: ["finance.*"]}
      device: {status: compliant, mdm: enrolled}
      app: {type: internal, group: "finance-apps"}
    checks: [dlp, data_encryption]

  - name: block-high-risk-saas
    action: block
    when:
      app: {risk: high, category: ["file_sharing", "shadow_it"]}
      user: {except: ["approvers@corp.com"]}
    notify: security-team

  - name: web-access-default
    action: inspect
    when:
      destination: {category: "general-internet"}
    checks: [malware_scan, url_reputation]

# policy_engine.py — 策略引擎
def evaluate_sase_policy(subject: dict, resource: dict, policies: list) -> dict:
    """按序匹配策略：命中即执行动作。"""
    for policy in policies:
        if matches(subject, resource, policy["when"]):
            return {
                "policy": policy["name"],
                "action": policy["action"],
                "checks": policy.get("checks", []),
            }
    return {"action": "block", "reason": "no-match-default-deny"}   # 默认拒绝

def build_default_deny(known_roles, known_apps):
    """任何未匹配策略的访问一律拒绝（零信任兜底）。"""
    def evaluate(subject, resource):
        if subject["role"] not in known_roles:
            return "block:unknown-role"
        if resource["id"] not in known_apps:
            return "block:unknown-app"
        return "allow"
    return evaluate

# dlp_classification.py
DATA_CLASSES = {
    "credit_card": r"\b(?:\d[ -]*?){13,16}\b",
    "cn_id":       r"\b\d{17}[\dXx]\b",
    "passport":    r"[A-Z]\d{7}",
    "source_code": r"(?s)(def |class |function )",
    "secret_key":  r"(?i)(api[_-]?key|secret|token)\s*[=:]\s*['\"][^'\"]+",
}

def classify_and_respond(content, destination, dlp_policy):
    detections = {k: bool(pattern.search(content))
                  for k, pattern in DATA_CLASSES.items()}
    actions = []
    for kind, detected in detections.items():
        if detected:
            action = dlp_policy.get(kind, "alert")
            actions.append(f"{action}:{kind} to {destination}")
    if any("block" in a for a in actions):
        return {"verdict": "block", "actions": actions}
    if actions:
        return {"verdict": "alert", "actions": actions}
    return {"verdict": "allow", "actions": []}

阶段 1（评估）：盘点 SaaS 应用、影子 IT、流量分布
阶段 2（试点）：选一条业务线开通 ZTNA + CASB，验证体验
阶段 3（扩展）：SWG 接管 Web 访问，FWaaS 替代部分硬件
阶段 4（整合）：SD-WAN 升级，策略统一编排
阶段 5（常态化）：持续 DLP、行为分析、策略优化

SASE 依赖强身份：
  企业 SSO → SASE 验证身份 → ZTNA 授权应用
  IAM（Entra ID / Okta）提供身份权威
  设备证书 + MDM 提供设备状态
  风险评分（Behavioral）动态调整访问

# 将 SASE 事件接入 SIEM
def forward_sase_events(sase_api, siem_client, poll_interval=60):
    """拉取 SASE 安全事件，标准化后送入 SIEM。"""
    while True:
        events = sase_api.get_events(cursor=last_cursor)
        for ev in events:
            siem_client.send(normalize_sase_event(ev))   # 统一格式
        sleep(poll_interval)

# 关键事件类型：
#  - blocked_access（拒绝访问）
#  - risky_behavior（风险行为）
#  - shadow_it_detected（影子应用）
#  - dlp_violation（数据泄漏）

def sase_dlp_reaction(sase_event, response_playbook):
    """根据 SASE 事件触发响应剧本。"""
    if sase_event["type"] == "dlp_violation":
        response_playbook.execute([
            revoke_user_access(sase_event["user"]),
            require_mfa_re_auth(sase_event["user"]),
            quarantine_file(sase_event["file_id"]),
            notify_dpo(),
        ])

def assess_sase_deployment(sase_metrics) -> dict:
    """评估 SASE 部署的安全与体验效果。"""
    return {
        # 安全效果
        "blocked_threats": sase_metrics["blocked_malware"],
        "shadow_it_reduced": pct_change(shadow_it_before, shadow_it_now),
        "dlp_violations": sase_metrics["dlp_count"],
        "lateral_exposure": sase_metrics["ztna_exposed_surface"],
        # 用户体验
        "avg_latency": sase_metrics["edge_latency_ms"],
        "vpn_incidents": sase_metrics["vpn_outages"],
    }

演练场景：
  1. 绕过 ZTNA：用非授权账号/设备访问私有应用 → 应被拒
  2. 数据外泄：向网盘上传含信用卡数据 → DLP 应拦截
  3. 恶意站点：访问已知钓鱼域名 → SWG 应阻断
  4. 影子应用：未批准 SaaS 的流量 → 应被发现/阻断