SASE Framework กับ Incident Management — วิธีใช้

SASE Framework คืออะไร

SASE (Secure Access Service Edge) เป็น Framework ที่รวม Network และ Security เข้าด้วยกัน ให้บริการจาก Cloud Edge ใกล้ผู้ใช้ ประกอบด้วย SD-WAN สำหรับ Network Optimization, SWG (Secure Web Gateway) สำหรับ Web Security, CASB (Cloud Access Security Broker) สำหรับ Cloud App Security, FWaaS (Firewall as a Service) และ ZTNA (Zero Trust Network Access)

การรวม SASE กับ Incident Management ช่วยให้เห็น Security Events จากทุก Layer ใน Dashboard เดียว Automate Response ได้เร็วขึ้น ลดเวลา MTTD (Mean Time to Detect) และ MTTR (Mean Time to Respond)

Component	หน้าที่	ตัวอย่าง
SD-WAN	Network Optimization	Intelligent Routing, QoS
SWG	Web Security	URL Filtering, Malware Protection
CASB	Cloud App Security	Shadow IT Detection, DLP
FWaaS	Firewall	IPS/IDS, App Control
ZTNA	Zero Trust Access	Identity-based Access, MFA

SASE Policy Configuration

=== SASE Policy Configuration ===

ตัวอย่าง Zero Trust Policy สำหรับ SASE

1. Cloudflare Zero Trust — Tunnel Config

cloudflared tunnel create my-tunnel

config.yml สำหรับ cloudflared

tunnel: abc123-def456

credentials-file: /etc/cloudflared/abc123-def456.json

ingress:

• hostname: app.example.com

service: http://localhost:8080

originRequest:

noTLSVerify: false

connectTimeout: 30s

• hostname: api.example.com

service: http://localhost:3000

originRequest:

httpHostHeader: api.internal

• service: http_status:404

2. Zero Trust Access Policy (Terraform)

resource "cloudflare_access_application" "internal_app" {

zone_id = var.zone_id

name = "Internal Application"

domain = "app.example.com"

type = "self_hosted"

เนื้อหาเกี่ยวข้อง — ทำความเข้าใจ hình bình hành có đường trung bình không

session_duration = "24h"

}

resource "cloudflare_access_policy" "allow_employees" {

application_id = cloudflare_access_application.internal_app.id

zone_id = var.zone_id

name = "Allow Employees"

แนะนำเพิ่มเติม — ติดตาม XM Signal

precedence = 1

decision = "allow"

include {

email_domain = ["example.com"]

}

require {

auth_method = "mfa"

}

}

3. Firewall Rules

resource "cloudflare_firewall_rule" "block_threats" {

zone_id = var.zone_id

description = "Block known threats"

filter_id = cloudflare_filter.threats.id

เนื้อหาเกี่ยวข้อง — อ่านต่อ: Qwik Resumability CDN Configuration

action = "block"

}

resource "cloudflare_filter" "threats" {

zone_id = var.zone_id

expression = "(cf.threat_score gt 14) or (cf.bot_management.score lt 30)"

}

4. DLP Policy

• ตรวจจับ Credit Card Numbers ใน Outbound Traffic
• Block Upload ไฟล์ที่มี PII ไป Unauthorized Cloud Apps
• Alert เมื่อพบ Source Code ถูก Upload ไป GitHub Personal

5. Network Segmentation

Segment: Engineering

• Access: GitHub, AWS Console, Internal Tools
• Deny: Social Media, Personal Email

Segment: Finance

• Access: Banking Apps, ERP, Internal Tools
• Deny: Developer Tools, SSH

Segment: Guest

แนะนำเพิ่มเติม — ระบบเทรดของ iCafeForex

• Access: Internet Only (Web Browsing)
• Deny: All Internal Resources

echo "SASE Policy Configuration Examples"

echo "==================================="

echo "1. Zero Trust Tunnel: cloudflared tunnel"

echo "2. Access Policy: Email Domain + MFA"

echo "3. Firewall: Threat Score + Bot Management"

echo "4. DLP: Credit Card, PII, Source Code"

echo "5. Segmentation: Engineering, Finance, Guest"

Incident Response Automation

# sase_incident.py — SASE Incident Response Automation
import json
import requests
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ThreatLevel(Enum):
    CRITICAL = 1
    HIGH = 2
    MEDIUM = 3
    LOW = 4
    INFO = 5

@dataclass
class SecurityEvent:
    timestamp: datetime
    source_ip: str
    user: str
    event_type: str
    threat_level: ThreatLevel
    description: str
    component: str  # SWG, CASB, FWaaS, ZTNA
    action_taken: str = ""
    resolved: bool = False

class SASEIncidentManager:
    """จัดการ Security Incidents จาก SASE Components"""

    def __init__(self):
        self.events: List[SecurityEvent] = []
        self.blocked_ips = set()
        self.suspended_users = set()
        self.playbooks = self._load_playbooks()

    def _load_playbooks(self):
        """Response Playbooks สำหรับ Event Types ต่างๆ"""
        return {
            "brute_force": {
                "actions": ["block_ip", "notify_soc", "lock_account"],
                "severity": ThreatLevel.HIGH,
                "auto_respond": True,
            },
            "malware_download": {
                "actions": ["block_url", "quarantine_file", "scan_device"],
                "severity": ThreatLevel.CRITICAL,
                "auto_respond": True,
            },
            "data_exfiltration": {
                "actions": ["block_upload", "notify_soc", "revoke_access"],
                "severity": ThreatLevel.CRITICAL,
                "auto_respond": True,
            },
            "unauthorized_app": {
                "actions": ["block_app", "notify_manager", "log_event"],
                "severity": ThreatLevel.MEDIUM,
                "auto_respond": False,
            },
            "policy_violation": {
                "actions": ["warn_user", "log_event", "notify_compliance"],
                "severity": ThreatLevel.LOW,
                "auto_respond": False,
            },
        }

    def ingest_event(self, event: SecurityEvent):
        """รับ Event และ Respond อัตโนมัติ"""
        self.events.append(event)
        logger.info(f"Event: [{event.threat_level.name}] "
                    f"{event.event_type} from {event.source_ip}")

        playbook = self.playbooks.get(event.event_type)
        if playbook and playbook["auto_respond"]:
            self._auto_respond(event, playbook)

    def _auto_respond(self, event, playbook):
        """ตอบสนองอัตโนมัติตาม Playbook"""
        actions_taken = []

        for action in playbook["actions"]:
            if action == "block_ip":
                self.blocked_ips.add(event.source_ip)
                actions_taken.append(f"Blocked IP: {event.source_ip}")

            elif action == "lock_account":
                self.suspended_users.add(event.user)
                actions_taken.append(f"Locked account: {event.user}")

            elif action == "revoke_access":
                self.suspended_users.add(event.user)
                actions_taken.append(f"Revoked access: {event.user}")

            elif action == "notify_soc":
                actions_taken.append("SOC notified via Slack/PagerDuty")

            else:
                actions_taken.append(f"Action: {action}")

        event.action_taken = "; ".join(actions_taken)
        event.resolved = True
        logger.info(f"Auto-response: {event.action_taken}")

    def dashboard(self):
        """แสดง Security Dashboard"""
        now = datetime.now()
        last_24h = [e for e in self.events
                    if e.timestamp > now - timedelta(hours=24)]

        by_level = {}
        by_component = {}
        for e in last_24h:
            by_level[e.threat_level.name] = by_level.get(e.threat_level.name, 0) + 1
            by_component[e.component] = by_component.get(e.component, 0) + 1

        print(f"\n{'='*55}")
        print(f"SASE Security Dashboard — {now:%Y-%m-%d %H:%M}")
        print(f"{'='*55}")
        print(f"  Events (24h): {len(last_24h)}")
        print(f"  Blocked IPs:  {len(self.blocked_ips)}")
        print(f"  Suspended:    {len(self.suspended_users)}")

        print(f"\n  By Threat Level:")
        for level, count in sorted(by_level.items()):
            print(f"    {level:<10} {count}")

        print(f"\n  By Component:")
        for comp, count in sorted(by_component.items()):
            print(f"    {comp:<10} {count}")

        # Recent Critical Events
        critical = [e for e in last_24h
                    if e.threat_level.value <= 2]
        if critical:
            print(f"\n  Recent Critical/High Events:")
            for e in critical[-5:]:
                print(f"    [{e.threat_level.name}] {e.event_type} "
                      f"— {e.user} ({e.source_ip})")

# ตัวอย่าง
mgr = SASEIncidentManager()

events = [
    SecurityEvent(datetime.now(), "203.0.113.50", "user1@example.com",
                  "brute_force", ThreatLevel.HIGH, "50 failed logins in 5 min", "ZTNA"),
    SecurityEvent(datetime.now(), "198.51.100.20", "user2@example.com",
                  "malware_download", ThreatLevel.CRITICAL, "Trojan detected", "SWG"),
    SecurityEvent(datetime.now(), "192.0.2.100", "user3@example.com",
                  "unauthorized_app", ThreatLevel.MEDIUM, "Dropbox Personal", "CASB"),
    SecurityEvent(datetime.now(), "10.0.0.50", "user4@example.com",
                  "data_exfiltration", ThreatLevel.CRITICAL, "Large upload to external", "CASB"),
]

for e in events:
    mgr.ingest_event(e)

mgr.dashboard()

Monitoring และ Alerting

=== SASE Monitoring Stack ===

docker-compose.yml สำหรับ SASE Monitoring

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: Calico Network Policy Stream Processing

version: '3.8'

services:

# Log Aggregation

elasticsearch:

image: elasticsearch:8.11.0

environment:

• discovery.type=single-node
• xpack.security.enabled=true

ports:

• "9200:9200"

volumes:

• es-data:/usr/share/elasticsearch/data

# Log Ingestion

logstash:

image: logstash:8.11.0

volumes:

• ./logstash/pipeline:/usr/share/logstash/pipeline

ports:

• "5044:5044"
• "5514:5514/udp"

# Visualization

kibana:

image: kibana:8.11.0

ports:

• "5601:5601"

environment:

• ELASTICSEARCH_HOSTS=http://elasticsearch:9200

# Alerting

elastalert:

image: jertel/elastalert2:latest

volumes:

• ./elastalert/rules:/opt/elastalert/rules

environment:

• ES_HOST=elasticsearch

volumes:

es-data:

=== ElastAlert Rules สำหรับ SASE Events ===

rules/brute_force.yaml

name: Brute Force Detection

type: frequency

index: sase-events-*

num_events: 10

timeframe:

minutes: 5

filter:

• term:

event_type: "authentication_failure"

alert:

• slack
• pagerduty

slack_webhook_url: "https://hooks.slack.com/services/xxx"

pagerduty_service_key: "xxx"

=== Prometheus Metrics สำหรับ SASE ===

เนื้อหาเกี่ยวข้อง — MLOps Pipeline Interview Preparation

prometheus.yml

scrape_configs:

• job_name: 'sase-metrics'

scrape_interval: 30s

static_configs:

• targets: ['sase-exporter:9090']

metrics_path: /metrics

Custom Metrics:

sase_events_total{component="SWG", severity="critical"} 5

sase_blocked_ips_total 42

sase_active_sessions 1500

sase_bandwidth_bytes{direction="inbound"} 1073741824

sase_policy_violations_total{policy="dlp"} 12

sase_mean_time_to_detect_seconds 45

sase_mean_time_to_respond_seconds 120

print("SASE Monitoring Stack:")

print(" - Elasticsearch: Log Storage & Search")

print(" - Logstash: Log Ingestion & Parsing")

print(" - Kibana: Visualization & Dashboards")

print(" - ElastAlert: Automated Alerting")

print(" - Prometheus: Metrics Collection")

print(" - Grafana: Metrics Visualization")

Best Practices

• Zero Trust: ตรวจสอบทุก Request ไม่ Trust อะไรโดยอัตโนมัติ ใช้ Identity-based Access
• Least Privilege: ให้สิทธิ์เท่าที่จำเป็น ทุก User ทุก Device ทุก Application
• Centralized Logging: รวม Logs จากทุก SASE Component ไว้ที่เดียว สำหรับ Investigation
• Automated Response: ใช้ SOAR Automate Response สำหรับ Events ที่รู้ Pattern
• Regular Assessment: ทบทวน Policy อย่างน้อยไตรมาสละครั้ง
• Phased Migration: Migrate จาก VPN เป็น SASE ทีละขั้น เริ่มจาก Pilot Group

SASE คืออะไร

Framework รวม Network (SD-WAN) และ Security (SWG CASB FWaaS ZTNA) เป็น Cloud-native Service ให้บริการจาก Edge ใกล้ผู้ใช้ ลด Latency เพิ่ม Security