SASE 구성요소 개념 다이어그램 (ASCII)
┌─────────────────────────────────────────────────────────────────────┐
│ SASE Architecture │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ Users/Devices SASE Cloud Platform │
│ ───────────── ─────────────────── │
│ │
│ [Remote Worker] ──────┐ ┌─────────────────────────────┐ │
│ │ │ Network Services │ │
│ [Branch Office] ──────┼────────►│ • SD-WAN │ │
│ │ │ • WAN Optimization │ │
│ [Mobile User] ────────┼─────────│ • Global PoP Network │ │
│ │ └─────────────────────────────┘ │
│ [IoT Device] ─────────┤ │ │
│ │ ▼ │
│ [HQ Office] ──────────┘ ┌─────────────────────────────┐ │
│ │ Security Services (SSE) │ │
│ │ • ZTNA (Zero Trust Access) │ │
│ │ • SWG (Web Gateway) │ │
│ │ • CASB (Cloud Security) │ │
│ │ • FWaaS (Cloud Firewall) │ │
│ │ • DLP (Data Loss Prevent) │ │
│ └─────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────┐ │
│ │ Destinations │ │
│ │ • SaaS (M365, Salesforce) │ │
│ │ • IaaS (AWS, Azure, GCP) │ │
│ │ • Private Apps (DC) │ │
│ │ • Internet │ │
│ └─────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
Terraform: Cloudflare Zero Trust (ZTNA) 설정
# providers.tf
terraform {
required_providers {
cloudflare = {
source = "cloudflare/cloudflare"
version = "~> 4.0"
}
}
}
provider "cloudflare" {
api_token = var.cloudflare_api_token
}
# variables.tf
variable "cloudflare_api_token" {
type = string
sensitive = true
}
variable "cloudflare_account_id" {
type = string
}
variable "tunnel_secret" {
type = string
sensitive = true
}
# Zero Trust Access Application 정의
resource "cloudflare_access_application" "internal_app" {
account_id = var.cloudflare_account_id
name = "Internal Dashboard"
domain = "dashboard.internal.example.com"
type = "self_hosted"
session_duration = "24h"
# 허용된 IdP 목록
allowed_idps = [cloudflare_access_identity_provider.okta.id]
# 자동 리다이렉트 비활성화 (API 접근용)
auto_redirect_to_identity = false
# CORS 설정
cors_headers {
allowed_methods = ["GET", "POST", "PUT", "DELETE"]
allowed_origins = ["https://dashboard.internal.example.com"]
allow_credentials = true
max_age = 600
}
}
# Access Policy - 누가 접근할 수 있는지 정의
resource "cloudflare_access_policy" "internal_app_policy" {
account_id = var.cloudflare_account_id
application_id = cloudflare_access_application.internal_app.id
name = "Allow Engineering Team"
precedence = 1
decision = "allow"
# 조건: 특정 그룹 + 기기 상태 확인
include {
group = [cloudflare_access_group.engineering.id]
}
require {
# 디바이스 포스처 확인 (필수)
device_posture = [cloudflare_device_posture_rule.corporate_device.id]
}
}
# Access Group 정의
resource "cloudflare_access_group" "engineering" {
account_id = var.cloudflare_account_id
name = "Engineering Team"
include {
email_domain = ["example.com"]
}
include {
okta {
name = "Engineering"
identity_provider_id = cloudflare_access_identity_provider.okta.id
}
}
}
# Device Posture Rule - 기기 보안 상태 확인
resource "cloudflare_device_posture_rule" "corporate_device" {
account_id = var.cloudflare_account_id
name = "Corporate Device Check"
type = "file"
match {
platform = "mac"
}
input {
id = "corporate_cert"
path = "/Library/Security/corporate-root-ca.pem"
exists = true
sha256 = "abc123..." # 인증서 해시
}
}
# Cloudflare Tunnel (private network 연결)
resource "cloudflare_tunnel" "internal_network" {
account_id = var.cloudflare_account_id
name = "internal-network-tunnel"
secret = base64encode(var.tunnel_secret)
}
resource "cloudflare_tunnel_config" "internal_network_config" {
account_id = var.cloudflare_account_id
tunnel_id = cloudflare_tunnel.internal_network.id
config {
ingress_rule {
hostname = "dashboard.internal.example.com"
service = "http://10.0.1.100:8080"
origin_request {
connect_timeout = "30s"
no_tls_verify = false
}
}
ingress_rule {
hostname = "api.internal.example.com"
service = "http://10.0.1.200:3000"
}
# Catch-all rule (required)
ingress_rule {
service = "http_status:404"
}
}
}
# Gateway Policy - SWG 웹 필터링
resource "cloudflare_teams_rule" "block_malware_sites" {
account_id = var.cloudflare_account_id
name = "Block Malware Sites"
description = "Block access to known malware domains"
precedence = 1
action = "block"
enabled = true
traffic = "any(dns.security_category[*] in {117 118 119 120})"
rule_settings {
block_page_enabled = true
block_page_reason = "This site has been blocked due to security risks."
}
}
# DLP Policy - 민감 데이터 유출 방지
resource "cloudflare_teams_rule" "dlp_credit_card" {
account_id = var.cloudflare_account_id
name = "Block Credit Card Data"
description = "Prevent credit card numbers from being uploaded"
precedence = 2
action = "block"
enabled = true
traffic = "http.request.body.mime == 'application/json' and any(http.request.body.json[*] matches '\\d{4}[- ]?\\d{4}[- ]?\\d{4}[- ]?\\d{4}')"
rule_settings {
block_page_enabled = true
block_page_reason = "Uploading credit card information is prohibited."
}
}
Python: SASE API 통합 및 정책 관리
"""
SASE 정책 관리 및 모니터링 스크립트
다양한 SASE 벤더 API와 통합하여 중앙 집중식 관리
"""
import requests
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
import json
@dataclass
class AccessPolicy:
"""접근 정책 데이터 모델"""
name: str
users: list[str]
groups: list[str]
applications: list[str]
conditions: dict
action: str # allow, block, isolate
class SASEProvider(ABC):
"""SASE 벤더 추상 클래스"""
@abstractmethod
def create_access_policy(self, policy: AccessPolicy) -> dict:
pass
@abstractmethod
def list_users(self) -> list[dict]:
pass
@abstractmethod
def get_security_events(self, hours: int = 24) -> list[dict]:
pass
class ZscalerClient(SASEProvider):
"""Zscaler ZPA/ZIA API 클라이언트"""
def __init__(self, client_id: str, client_secret: str, cloud: str = "zscaler"):
self.base_url = f"https://config.{cloud}.net/api/v1"
self.client_id = client_id
self.client_secret = client_secret
self.token = None
def _authenticate(self):
"""API 인증"""
response = requests.post(
f"{self.base_url}/auth/login",
json={
"clientId": self.client_id,
"clientSecret": self.client_secret
}
)
response.raise_for_status()
self.token = response.json().get("token")
def _headers(self):
if not self.token:
self._authenticate()
return {
"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json"
}
def create_access_policy(self, policy: AccessPolicy) -> dict:
"""ZPA 접근 정책 생성"""
payload = {
"name": policy.name,
"rule_order": "1",
"action": policy.action.upper(),
"conditions": [
{
"operands": [
{"object_type": "APP", "values": policy.applications}
]
},
{
"operands": [
{"object_type": "SCIM_GROUP", "values": policy.groups}
]
}
]
}
response = requests.post(
f"{self.base_url}/policySet/rules",
headers=self._headers(),
json=payload
)
response.raise_for_status()
return response.json()
def list_users(self) -> list[dict]:
"""사용자 목록 조회"""
response = requests.get(
f"{self.base_url}/users",
headers=self._headers()
)
response.raise_for_status()
return response.json().get("users", [])
def get_security_events(self, hours: int = 24) -> list[dict]:
"""보안 이벤트 조회 (ZIA)"""
response = requests.get(
f"{self.base_url}/security/events",
headers=self._headers(),
params={"hours": hours}
)
response.raise_for_status()
return response.json().get("events", [])
class CloudflareZeroTrustClient(SASEProvider):
"""Cloudflare Zero Trust API 클라이언트"""
def __init__(self, api_token: str, account_id: str):
self.base_url = f"https://api.cloudflare.com/client/v4/accounts/{account_id}"
self.headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json"
}
def create_access_policy(self, policy: AccessPolicy) -> dict:
"""Access Policy 생성"""
# 먼저 Application 확인 필요
payload = {
"name": policy.name,
"decision": policy.action,
"include": [
{"group": {"id": g}} for g in policy.groups
],
"require": []
}
if policy.conditions.get("device_posture"):
payload["require"].append({
"device_posture": {"integration_uid": policy.conditions["device_posture"]}
})
response = requests.post(
f"{self.base_url}/access/apps/{policy.applications[0]}/policies",
headers=self.headers,
json=payload
)
response.raise_for_status()
return response.json().get("result", {})
def list_users(self) -> list[dict]:
"""Access 사용자 목록"""
response = requests.get(
f"{self.base_url}/access/users",
headers=self.headers
)
response.raise_for_status()
return response.json().get("result", [])
def get_security_events(self, hours: int = 24) -> list[dict]:
"""Gateway 로그 조회"""
response = requests.get(
f"{self.base_url}/gateway/audit_logs",
headers=self.headers,
params={"per_page": 1000}
)
response.raise_for_status()
return response.json().get("result", [])
def create_gateway_rule(self, name: str, traffic_filter: str, action: str):
"""Gateway 정책 생성 (SWG/DNS 필터링)"""
payload = {
"name": name,
"enabled": True,
"action": action,
"traffic": traffic_filter,
"rule_settings": {
"block_page_enabled": True
}
}
response = requests.post(
f"{self.base_url}/gateway/rules",
headers=self.headers,
json=payload
)
response.raise_for_status()
return response.json().get("result", {})
class SASEPolicyManager:
"""멀티 벤더 SASE 정책 통합 관리"""
def __init__(self):
self.providers: dict[str, SASEProvider] = {}
def register_provider(self, name: str, provider: SASEProvider):
self.providers[name] = provider
def deploy_policy_to_all(self, policy: AccessPolicy):
"""모든 SASE 플랫폼에 동일 정책 배포"""
results = {}
for name, provider in self.providers.items():
try:
result = provider.create_access_policy(policy)
results[name] = {"status": "success", "data": result}
except Exception as e:
results[name] = {"status": "error", "message": str(e)}
return results
def aggregate_security_events(self, hours: int = 24) -> list[dict]:
"""모든 플랫폼의 보안 이벤트 통합"""
all_events = []
for name, provider in self.providers.items():
events = provider.get_security_events(hours)
for event in events:
event["source_provider"] = name
all_events.extend(events)
return sorted(all_events, key=lambda x: x.get("timestamp", ""), reverse=True)
# 사용 예시
if __name__ == "__main__":
# 매니저 초기화
manager = SASEPolicyManager()
# Cloudflare 등록
cf_client = CloudflareZeroTrustClient(
api_token="your-api-token",
account_id="your-account-id"
)
manager.register_provider("cloudflare", cf_client)
# 정책 정의
engineering_access = AccessPolicy(
name="Engineering Internal Access",
users=["user1@example.com", "user2@example.com"],
groups=["engineering-team"],
applications=["internal-dashboard-app-id"],
conditions={
"device_posture": "corporate-device-rule-id",
"location": ["US", "KR"]
},
action="allow"
)
# 정책 배포
results = manager.deploy_policy_to_all(engineering_access)
print(json.dumps(results, indent=2))
# 보안 이벤트 조회
events = manager.aggregate_security_events(hours=24)
print(f"Total security events: {len(events)}")