SASE Security Compliance Automation คืออะไร
SASE (Secure Access Service Edge) ต้องปฏิบัติตาม compliance standards หลายมาตรฐาน เช่น GDPR, PCI DSS, PDPA, SOC 2, ISO 27001 และ HIPAA Compliance Automation คือการใช้เครื่องมือและ scripts อัตโนมัติเพื่อตรวจสอบ รายงาน และบังคับใช้ compliance policies แทนการทำ manual ช่วยลดเวลา audit จากสัปดาห์เหลือชั่วโมง ลด human error และทำให้ compliance เป็น continuous process แทนที่จะเป็น periodic event บทความนี้อธิบายการสร้าง compliance automation สำหรับ SASE security ด้วย Python ครบทุกมาตรฐาน
Compliance Frameworks
# frameworks.py — Compliance frameworks for SASE
import json
class ComplianceFrameworks:
FRAMEWORKS = {
"gdpr": {
"name": "GDPR (EU General Data Protection Regulation)",
"sase_requirements": [
"Data Processing Records — log ทุกการ access ข้อมูลส่วนบุคคล",
"DLP Enforcement — ป้องกัน data leak ผ่าน CASB/SWG",
"Right to Access — ให้ผู้ใช้เข้าถึง logs ของตัวเอง",
"Data Minimization — เก็บ logs เท่าที่จำเป็น",
"Breach Notification — แจ้งภายใน 72 ชั่วโมง",
],
},
"pci_dss": {
"name": "PCI DSS (Payment Card Industry)",
"sase_requirements": [
"Firewall Rules Review — ตรวจสอบ FWaaS rules ทุก 6 เดือน",
"Encryption — SSL/TLS inspection สำหรับ cardholder data",
"Access Control — ZTNA least-privilege access",
"Log Retention — เก็บ logs อย่างน้อย 1 ปี",
"Vulnerability Management — scan ทุกไตรมาส",
],
},
"pdpa": {
"name": "PDPA (Thailand Personal Data Protection Act)",
"sase_requirements": [
"Consent Management — บันทึกการยินยอม",
"Data Protection — encrypt ข้อมูลส่วนบุคคล",
"Access Logging — log การเข้าถึงข้อมูล",
"DPO Notification — แจ้ง DPO เมื่อมี breach",
"Cross-border Transfer — ควบคุมการส่งข้อมูลข้ามประเทศ",
],
},
"soc2": {
"name": "SOC 2 (Service Organization Control)",
"sase_requirements": [
"Access Controls — ZTNA + MFA enforcement",
"Change Management — audit trail สำหรับ policy changes",
"Monitoring — continuous monitoring + alerting",
"Incident Response — documented IR procedures",
"Risk Assessment — periodic risk evaluation",
],
},
}
def show_frameworks(self):
print("=== Compliance Frameworks ===\n")
for key, fw in self.FRAMEWORKS.items():
print(f"[{fw['name']}]")
for req in fw["sase_requirements"][:3]:
print(f" • {req}")
print()
frameworks = ComplianceFrameworks()
frameworks.show_frameworks()Compliance Checker Script
# checker.py — Automated compliance checker
import json
class ComplianceChecker:
CODE = """
# compliance_checker.py — Check SASE compliance automatically
import requests
import json
from datetime import datetime, timedelta
class SASEComplianceChecker:
def __init__(self, sase_api_url, api_token):
self.api_url = sase_api_url
self.headers = {"Authorization": f"Bearer {api_token}"}
self.results = []
def check_all(self):
'''Run all compliance checks'''
checks = [
self.check_encryption,
self.check_access_controls,
self.check_dlp_policies,
self.check_log_retention,
self.check_mfa_enforcement,
self.check_firewall_rules,
self.check_incident_response,
]
for check in checks:
result = check()
self.results.append(result)
status = "PASS" if result["pass"] else "FAIL"
print(f" [{status:>4}] {result['name']}")
score = sum(1 for r in self.results if r["pass"]) / len(self.results) * 100
return {"score": round(score, 1), "results": self.results}
def check_encryption(self):
'''Check SSL/TLS inspection is enabled'''
resp = requests.get(
f"{self.api_url}/policies/ssl-inspection",
headers=self.headers
)
config = resp.json()
enabled = config.get("enabled", False)
return {
"name": "SSL/TLS Inspection",
"pass": enabled,
"detail": "SSL inspection enabled" if enabled else "SSL inspection DISABLED",
"framework": ["PCI DSS 4.1", "SOC 2"],
}
def check_access_controls(self):
'''Check ZTNA policies exist and are active'''
resp = requests.get(
f"{self.api_url}/ztna/policies",
headers=self.headers
)
policies = resp.json().get("data", [])
active = [p for p in policies if p.get("status") == "active"]
return {
"name": "ZTNA Access Controls",
"pass": len(active) > 0,
"detail": f"{len(active)} active ZTNA policies",
"framework": ["SOC 2 CC6.1", "PCI DSS 7.1"],
}
def check_dlp_policies(self):
'''Check DLP policies for personal data protection'''
resp = requests.get(
f"{self.api_url}/casb/dlp-policies",
headers=self.headers
)
policies = resp.json().get("data", [])
has_pii = any("PII" in p.get("name", "") or "personal" in p.get("name", "").lower() for p in policies)
return {
"name": "DLP Policies (PII Protection)",
"pass": has_pii,
"detail": f"{len(policies)} DLP policies, PII protection: {'Yes' if has_pii else 'No'}",
"framework": ["GDPR Art.32", "PDPA"],
}
def check_log_retention(self):
'''Check log retention meets requirements'''
resp = requests.get(
f"{self.api_url}/settings/log-retention",
headers=self.headers
)
retention_days = resp.json().get("retention_days", 0)
return {
"name": "Log Retention (>= 365 days)",
"pass": retention_days >= 365,
"detail": f"Retention: {retention_days} days",
"framework": ["PCI DSS 10.7"],
}
def check_mfa_enforcement(self):
'''Check MFA is enforced for all users'''
resp = requests.get(
f"{self.api_url}/settings/mfa",
headers=self.headers
)
enforced = resp.json().get("enforced", False)
return {
"name": "MFA Enforcement",
"pass": enforced,
"detail": "MFA enforced" if enforced else "MFA NOT enforced",
"framework": ["SOC 2 CC6.1", "PCI DSS 8.3"],
}
def check_firewall_rules(self):
'''Check firewall rules have been reviewed recently'''
resp = requests.get(
f"{self.api_url}/firewall/rules",
headers=self.headers
)
rules = resp.json().get("data", [])
stale = [r for r in rules if not r.get("last_reviewed") or
(datetime.now() - datetime.fromisoformat(r["last_reviewed"])).days > 180]
return {
"name": "Firewall Rules Review (< 180 days)",
"pass": len(stale) == 0,
"detail": f"{len(stale)} stale rules out of {len(rules)}",
"framework": ["PCI DSS 1.1.7"],
}
def check_incident_response(self):
'''Check IR plan exists and was tested'''
resp = requests.get(
f"{self.api_url}/incident-response/plan",
headers=self.headers
)
plan = resp.json()
has_plan = plan.get("exists", False)
tested = plan.get("last_test_date")
recently_tested = False
if tested:
recently_tested = (datetime.now() - datetime.fromisoformat(tested)).days < 365
return {
"name": "Incident Response Plan",
"pass": has_plan and recently_tested,
"detail": f"Plan: {'Yes' if has_plan else 'No'}, Tested: {'Yes' if recently_tested else 'No'}",
"framework": ["SOC 2 CC7.4", "PCI DSS 12.10"],
}
checker = SASEComplianceChecker(api_url, token)
report = checker.check_all()
print(f"\\nCompliance Score: {report['score']}%")
"""
def show_code(self):
print("=== Compliance Checker ===")
print(self.CODE[:600])
checker = ComplianceChecker()
checker.show_code()Automated Reporting
# reporting.py — Automated compliance reporting
import json
import random
class ComplianceReporting:
CODE = """
# report_generator.py — Generate compliance reports
import json
from datetime import datetime
from jinja2 import Template
class ComplianceReportGenerator:
def __init__(self, checker):
self.checker = checker
def generate_html_report(self, output_path="compliance_report.html"):
results = self.checker.check_all()
html_template = Template('''
SASE Compliance Report
Generated: {{ timestamp }}
Score: {{ score }}%
Check Status Detail Frameworks
{{ r.name }}
{{ "PASS" if r.pass else "FAIL" }}
{{ r.detail }}
{{ r.framework | join(", ") }}
{% endfor %}
''')
html = html_template.render(
timestamp=datetime.utcnow().isoformat(),
score=results["score"],
results=results["results"],
)
with open(output_path, 'w') as f:
f.write(html)
return output_path
def generate_json_report(self):
results = self.checker.check_all()
return json.dumps(results, indent=2, default=str)
def send_to_slack(self, webhook_url, results):
import requests
score = results["score"]
failed = [r for r in results["results"] if not r["pass"]]
message = f"📋 SASE Compliance Score: {score}%\\n"
if failed:
message += f"⚠️ {len(failed)} checks failed:\\n"
for f in failed:
message += f" • {f['name']}: {f['detail']}\\n"
else:
message += "✅ All checks passed!"
requests.post(webhook_url, json={"text": message})
"""
def show_code(self):
print("=== Report Generator ===")
print(self.CODE[:600])
def sample_report(self):
print(f"\n=== Compliance Report Summary ===")
checks = [
{"name": "SSL/TLS Inspection", "pass": True},
{"name": "ZTNA Access Controls", "pass": True},
{"name": "DLP Policies (PII)", "pass": random.random() > 0.2},
{"name": "Log Retention >= 365d", "pass": True},
{"name": "MFA Enforcement", "pass": True},
{"name": "Firewall Rules Review", "pass": random.random() > 0.3},
{"name": "Incident Response Plan", "pass": random.random() > 0.2},
]
passed = sum(1 for c in checks if c["pass"])
score = passed / len(checks) * 100
for c in checks:
status = "PASS" if c["pass"] else "FAIL"
print(f" [{status:>4}] {c['name']}")
print(f"\n Score: {score:.0f}% ({passed}/{len(checks)} checks passed)")
reporting = ComplianceReporting()
reporting.show_code()
reporting.sample_report()
''')
html = html_template.render(
timestamp=datetime.utcnow().isoformat(),
score=results["score"],
results=results["results"],
)
with open(output_path, 'w') as f:
f.write(html)
return output_path
def generate_json_report(self):
results = self.checker.check_all()
return json.dumps(results, indent=2, default=str)
def send_to_slack(self, webhook_url, results):
import requests
score = results["score"]
failed = [r for r in results["results"] if not r["pass"]]
message = f"📋 SASE Compliance Score: {score}%\\n"
if failed:
message += f"⚠️ {len(failed)} checks failed:\\n"
for f in failed:
message += f" • {f['name']}: {f['detail']}\\n"
else:
message += "✅ All checks passed!"
requests.post(webhook_url, json={"text": message})
"""
def show_code(self):
print("=== Report Generator ===")
print(self.CODE[:600])
def sample_report(self):
print(f"\n=== Compliance Report Summary ===")
checks = [
{"name": "SSL/TLS Inspection", "pass": True},
{"name": "ZTNA Access Controls", "pass": True},
{"name": "DLP Policies (PII)", "pass": random.random() > 0.2},
{"name": "Log Retention >= 365d", "pass": True},
{"name": "MFA Enforcement", "pass": True},
{"name": "Firewall Rules Review", "pass": random.random() > 0.3},
{"name": "Incident Response Plan", "pass": random.random() > 0.2},
]
passed = sum(1 for c in checks if c["pass"])
score = passed / len(checks) * 100
for c in checks:
status = "PASS" if c["pass"] else "FAIL"
print(f" [{status:>4}] {c['name']}")
print(f"\n Score: {score:.0f}% ({passed}/{len(checks)} checks passed)")
reporting = ComplianceReporting()
reporting.show_code()
reporting.sample_report()Policy-as-Code
# policy_as_code.py — OPA/Rego for SASE compliance
import json
class PolicyAsCode:
OPA_POLICIES = """
# sase_compliance.rego — OPA policies for SASE compliance
package sase.compliance
# Check: All ZTNA policies must require MFA for production
deny[msg] {
input.ztna_policies[i].environment == "production"
not input.ztna_policies[i].require_mfa
msg := sprintf("ZTNA policy '%s' must require MFA for production", [input.ztna_policies[i].name])
}
# Check: DLP must be enabled for all CASB policies
deny[msg] {
input.casb_policies[i].type == "sanctioned_app"
not input.casb_policies[i].dlp_enabled
msg := sprintf("CASB policy '%s' must have DLP enabled", [input.casb_policies[i].name])
}
# Check: Firewall default action must be deny
deny[msg] {
input.firewall.default_action != "deny"
msg := "Firewall default action must be 'deny' (deny-all approach)"
}
# Check: SSL inspection must be enabled
deny[msg] {
not input.ssl_inspection.enabled
msg := "SSL inspection must be enabled for compliance"
}
# Check: Log retention minimum 365 days
deny[msg] {
input.log_retention_days < 365
msg := sprintf("Log retention must be >= 365 days (current: %d)", [input.log_retention_days])
}
# Compliance score
score = s {
total := count(deny)
checks := 5 # Total number of checks
s := (checks - total) / checks * 100
}
"""
TERRAFORM = """
# compliance_checks.tf — Terraform compliance checks
resource "null_resource" "compliance_check" {
triggers = {
always_run = timestamp()
}
provisioner "local-exec" {
command = <<-EOT
python3 scripts/compliance_checker.py \\
--api-url $SSE_API_URL \\
--api-token $SSE_API_TOKEN \\
--output reports/compliance_$(date +%Y%m%d).json \\
--fail-on-score-below 80
EOT
}
}
# Schedule compliance check via cron
resource "aws_cloudwatch_event_rule" "compliance_daily" {
name = "sase-compliance-daily"
schedule_expression = "cron(0 6 * * ? *)" # Daily 6 AM UTC
}
"""
def show_opa(self):
print("=== OPA Compliance Policies ===")
print(self.OPA_POLICIES[:500])
def show_terraform(self):
print(f"\n=== Terraform ===")
print(self.TERRAFORM[:400])
pac = PolicyAsCode()
pac.show_opa()
pac.show_terraform()CI/CD Compliance Pipeline
# cicd.py — CI/CD pipeline for compliance automation
import json
class CompliancePipeline:
GITHUB_ACTIONS = """
# .github/workflows/compliance.yml
name: SASE Compliance Check
on:
schedule:
- cron: '0 6 * * 1' # Every Monday 6 AM UTC
workflow_dispatch:
jobs:
compliance-check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
- run: pip install requests jinja2 pyyaml
- name: Run compliance checks
run: python scripts/compliance_checker.py --output report.json
env:
SSE_API_URL: }
SSE_API_TOKEN: }
- name: Generate HTML report
run: python scripts/generate_report.py report.json --html report.html
- name: Upload report
uses: actions/upload-artifact@v4
with:
name: compliance-report
path: report.html
- name: Notify Slack
if: always()
run: python scripts/notify_slack.py report.json
env:
SLACK_WEBHOOK: }
- name: Fail if score below threshold
run: |
score=$(python -c "import json; print(json.load(open('report.json'))['score'])")
if (( $(echo "$score < 80" | bc -l) )); then
echo "Compliance score $score% is below 80% threshold"
exit 1
fi
"""
def show_pipeline(self):
print("=== CI/CD Pipeline ===")
print(self.GITHUB_ACTIONS[:500])
def schedule(self):
print(f"\n=== Compliance Schedule ===")
schedule = [
{"frequency": "Daily", "task": "Automated log review + anomaly alerts"},
{"frequency": "Weekly", "task": "Full compliance check + report"},
{"frequency": "Monthly", "task": "Policy review + update recommendations"},
{"frequency": "Quarterly", "task": "Vulnerability scan + risk assessment"},
{"frequency": "Annually", "task": "Full audit + IR plan test + cert renewal"},
]
for s in schedule:
print(f" [{s['frequency']:<12}] {s['task']}")
pipeline = CompliancePipeline()
pipeline.show_pipeline()
pipeline.schedule()FAQ - คำถามที่พบบ่อย
Q: Compliance automation คุ้มค่าไหม?
A: คุ้มมาก: Manual audit: 2-4 สัปดาห์ × 2-4 คน = 160-640 man-hours Automated: setup 40-80 hours → ทำงานอัตโนมัติทุกสัปดาห์ ROI: คืนทุนภายใน 1-2 audit cycles ลด human error, consistent results, continuous compliance แทน periodic
Q: เริ่มต้น compliance automation จากไหน?
A: Step 1: ระบุ frameworks ที่ต้อง comply (GDPR, PCI DSS, PDPA) Step 2: Map SASE controls กับ compliance requirements Step 3: สร้าง automated checks สำหรับ requirements ที่ตรวจได้อัตโนมัติ Step 4: สร้าง reporting + alerting Step 5: Integrate กับ CI/CD (weekly automated checks) เริ่มจาก checks ที่ง่าย (MFA, encryption, log retention) แล้วค่อยเพิ่ม
Q: Policy-as-Code จำเป็นไหม?
A: แนะนำอย่างยิ่ง: Version control — track ทุก policy change, rollback ได้ Code review — ทุก policy change ผ่าน PR review Automated testing — validate policies ก่อน apply Audit trail — git log = compliance evidence Tools: OPA/Rego (general), Sentinel (HashiCorp), custom Python scripts
Q: PDPA ต่างจาก GDPR อย่างไร?
A: คล้ายกันมาก — PDPA ได้รับอิทธิพลจาก GDPR: ทั้งคู่: consent-based, data subject rights, breach notification, DPO ต่างกัน: PDPA มี grace period สำหรับ existing data, fine structure ต่างกัน, enforcement อาจต่างกัน SASE automation: สร้าง checks ที่ครอบคลุมทั้ง GDPR + PDPA — overlap มาก ใช้ script เดียวกันได้ ปรับ thresholds/rules ตาม jurisdiction