Burp Suite Pro Edge Computing

Burp Suite Pro และ Edge Computing Security

Burp Suite Pro เป็นเครื่องมือมาตรฐานอุตสาหกรรมสำหรับทดสอบความปลอดภัยเว็บแอปพลิเคชัน เมื่อนำมาใช้กับ Edge Computing Applications ต้องคำนึงถึงความเสี่ยงเฉพาะ เช่น Distributed Architecture, Multi-region Endpoints และ Edge-specific APIs

Edge Computing ประมวลผลข้อมูลใกล้ผู้ใช้ ลด Latency แต่เพิ่ม Attack Surface เพราะมีหลาย Edge Locations ที่ต้องรักษาความปลอดภัย การทดสอบด้วย Burp Suite ช่วยหาช่องโหว่ก่อนที่ Attacker จะค้นพบ

Setup Burp Suite สำหรับ Edge Testing

# === Burp Suite Pro Setup สำหรับ Edge Computing Testing ===

# 1. ติดตั้ง Burp Suite Pro
# ดาวน์โหลดจาก https://portswigger.net/burp/pro
# ต้องมี License Key

# 2. ตั้งค่า Proxy
# Burp Proxy: 127.0.0.1:8080
# Browser ตั้ง Proxy ชี้ไปที่ 127.0.0.1:8080
# ติดตั้ง Burp CA Certificate สำหรับ HTTPS Interception

# 3. ตั้งค่า Scope สำหรับ Edge Endpoints
# Target > Scope > Add
# Include:
#   *.edge.example.com
#   *.cdn.example.com
#   api.example.com/edge/*
#   *.workers.dev
#   *.cloudfunctions.net

# 4. Configure Session Handling
# Project Options > Sessions > Session Handling Rules
# Add Rule: Check Session Is Valid
# - Check response for: "401" or "token_expired"
# - If invalid: Run macro to refresh token

# 5. Edge-specific Headers
# Proxy > Options > Match and Replace
# Add rules to test different Edge behaviors:
#   Add Header: X-Forwarded-For: 203.0.113.1
#   Add Header: CF-Connecting-IP: 203.0.113.1
#   Add Header: X-Real-IP: 203.0.113.1
#   Add Header: X-Edge-Location: ap-southeast-1

# 6. Configure for HTTP/2
# Project Options > HTTP > HTTP/2
# Enable HTTP/2 for target hosts

# 7. Import OpenAPI Spec
# Target > Site Map > Right-click > Import > OpenAPI
# Upload edge-api-spec.yaml
# This maps all API endpoints automatically

# 8. Extension: Edge Security Scanner (Python)
# Extender > Extensions > Add
# Type: Python
# File: edge_scanner.py (see below)

Python Extension สำหรับ Burp Suite

# edge_scanner.py — Burp Suite Extension สำหรับ Edge Security
# ใช้กับ Burp Extender API (Jython/Python)

from burp import IBurpExtender, IScannerCheck, IScanIssue
from java.net import URL
import re
import json

class BurpExtender(IBurpExtender, IScannerCheck):
    """Burp Extension ตรวจสอบ Edge Computing Vulnerabilities"""

    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helpers = callbacks.getHelpers()
        callbacks.setExtensionName("Edge Security Scanner")
        callbacks.registerScannerCheck(self)
        print("[*] Edge Security Scanner loaded")

    def doPassiveScan(self, baseRequestResponse):
        """Passive Scan — ตรวจสอบ Response Headers"""
        issues = []
        response = baseRequestResponse.getResponse()
        if response is None:
            return None

        analyzedResponse = self._helpers.analyzeResponse(response)
        headers = analyzedResponse.getHeaders()

        # Check 1: Edge Cache Headers
        cache_issues = self._check_cache_headers(headers,
                                                  baseRequestResponse)
        issues.extend(cache_issues)

        # Check 2: Security Headers
        security_issues = self._check_security_headers(headers,
                                                        baseRequestResponse)
        issues.extend(security_issues)

        # Check 3: Edge Location Disclosure
        location_issues = self._check_edge_location(headers,
                                                     baseRequestResponse)
        issues.extend(location_issues)

        return issues if issues else None

    def _check_cache_headers(self, headers, reqResp):
        """ตรวจสอบ Cache Headers ที่อาจ Cache ข้อมูลสำคัญ"""
        issues = []
        header_str = "\n".join(headers).lower()

        # ตรวจสอบว่า Response ที่มีข้อมูลส่วนตัวถูก Cache
        body = self._helpers.bytesToString(reqResp.getResponse())

        sensitive_patterns = [
            r'"email"\s*:', r'"password"\s*:', r'"token"\s*:',
            r'"api_key"\s*:', r'"secret"\s*:',
        ]

        has_sensitive = any(re.search(p, body) for p in sensitive_patterns)
        is_cached = "cache-control: public" in header_str or \
                    "x-cache: hit" in header_str

        if has_sensitive and is_cached:
            issues.append(CustomScanIssue(
                reqResp.getHttpService(),
                self._helpers.analyzeRequest(reqResp).getUrl(),
                [reqResp],
                "Edge Cache Stores Sensitive Data",
                "Response containing sensitive data is being cached "
                "at the Edge. This could expose user data to other users.",
                "High",
            ))

        return issues

    def _check_security_headers(self, headers, reqResp):
        """ตรวจสอบ Security Headers ที่ขาดหาย"""
        issues = []
        header_str = "\n".join(headers).lower()

        required_headers = {
            "strict-transport-security": "HSTS",
            "x-content-type-options": "X-Content-Type-Options",
            "x-frame-options": "X-Frame-Options",
            "content-security-policy": "CSP",
        }

        missing = []
        for header, name in required_headers.items():
            if header not in header_str:
                missing.append(name)

        if missing:
            issues.append(CustomScanIssue(
                reqResp.getHttpService(),
                self._helpers.analyzeRequest(reqResp).getUrl(),
                [reqResp],
                "Missing Security Headers at Edge",
                "Missing headers: " + ", ".join(missing),
                "Medium",
            ))

        return issues

    def _check_edge_location(self, headers, reqResp):
        """ตรวจสอบ Edge Location Information Disclosure"""
        issues = []
        edge_headers = [
            "x-served-by", "x-cache", "cf-ray", "x-amz-cf-id",
            "x-edge-location", "server-timing",
        ]

        disclosed = []
        for h in headers:
            for eh in edge_headers:
                if h.lower().startswith(eh):
                    disclosed.append(h)

        if len(disclosed) > 3:
            issues.append(CustomScanIssue(
                reqResp.getHttpService(),
                self._helpers.analyzeRequest(reqResp).getUrl(),
                [reqResp],
                "Edge Infrastructure Information Disclosure",
                "Multiple edge headers expose infrastructure details: "
                + ", ".join(disclosed[:5]),
                "Low",
            ))

        return issues

    def doActiveScan(self, baseRequestResponse, insertionPoint):
        return None

    def consolidateDuplicateIssues(self, existingIssue, newIssue):
        if existingIssue.getIssueName() == newIssue.getIssueName():
            return -1
        return 0

class CustomScanIssue(IScanIssue):
    """Custom Scan Issue"""
    def __init__(self, httpService, url, reqResps, name, detail, severity):
        self._httpService = httpService
        self._url = url
        self._reqResps = reqResps
        self._name = name
        self._detail = detail
        self._severity = severity

    def getUrl(self): return self._url
    def getIssueName(self): return self._name
    def getIssueType(self): return 0
    def getSeverity(self): return self._severity
    def getConfidence(self): return "Certain"
    def getIssueBackground(self): return None
    def getRemediationBackground(self): return None
    def getIssueDetail(self): return self._detail
    def getRemediationDetail(self): return None
    def getHttpMessages(self): return self._reqResps
    def getHttpService(self): return self._httpService

Automated Edge Security Testing

# edge_security_test.py — Automated Security Testing สำหรับ Edge APIs
import requests
import json
from urllib.parse import urljoin

class EdgeSecurityTester:
    """ทดสอบ Security ของ Edge APIs"""

    def __init__(self, base_url, auth_token=None):
        self.base_url = base_url
        self.session = requests.Session()
        if auth_token:
            self.session.headers["Authorization"] = f"Bearer {auth_token}"
        self.results = []

    def test_cors(self, endpoint="/api/data"):
        """ทดสอบ CORS Misconfiguration"""
        url = urljoin(self.base_url, endpoint)
        evil_origins = [
            "https://evil.com",
            "https://attacker.example.com",
            "null",
        ]

        for origin in evil_origins:
            resp = self.session.get(url, headers={"Origin": origin})
            acao = resp.headers.get("Access-Control-Allow-Origin", "")

            if acao == origin or acao == "*":
                self.results.append({
                    "test": "CORS", "severity": "HIGH",
                    "detail": f"Reflects origin: {origin} -> {acao}",
                })
                print(f"  [HIGH] CORS reflects: {origin}")

    def test_cache_poisoning(self, endpoint="/"):
        """ทดสอบ Cache Poisoning"""
        url = urljoin(self.base_url, endpoint)
        payloads = [
            {"X-Forwarded-Host": "evil.com"},
            {"X-Original-URL": "/admin"},
            {"X-Rewrite-URL": "/admin"},
        ]

        for headers in payloads:
            resp = self.session.get(url, headers=headers)
            body = resp.text.lower()

            for key, value in headers.items():
                if value.lower() in body:
                    self.results.append({
                        "test": "Cache Poisoning", "severity": "HIGH",
                        "detail": f"Header {key}:{value} reflected in response",
                    })
                    print(f"  [HIGH] Cache Poisoning via {key}")

    def test_rate_limiting(self, endpoint="/api/login", requests_count=100):
        """ทดสอบ Rate Limiting"""
        url = urljoin(self.base_url, endpoint)
        blocked = False

        for i in range(requests_count):
            resp = self.session.post(url, json={
                "username": "test", "password": "wrong"
            })
            if resp.status_code == 429:
                blocked = True
                print(f"  [INFO] Rate limited at request {i+1}")
                break

        if not blocked:
            self.results.append({
                "test": "Rate Limiting", "severity": "MEDIUM",
                "detail": f"No rate limiting after {requests_count} requests",
            })
            print(f"  [MEDIUM] No rate limiting detected")

    def test_edge_bypass(self, endpoint="/admin"):
        """ทดสอบ Edge WAF/ACL Bypass"""
        url = urljoin(self.base_url, endpoint)
        bypass_techniques = [
            {"path": endpoint, "headers": {}},
            {"path": endpoint.upper(), "headers": {}},
            {"path": f"{endpoint}/.", "headers": {}},
            {"path": f"{endpoint}%2f", "headers": {}},
            {"path": endpoint, "headers": {"X-Original-URL": endpoint}},
            {"path": endpoint, "headers": {"X-Forwarded-For": "127.0.0.1"}},
        ]

        for technique in bypass_techniques:
            test_url = urljoin(self.base_url, technique["path"])
            resp = self.session.get(test_url, headers=technique["headers"],
                                   allow_redirects=False)
            if resp.status_code == 200:
                self.results.append({
                    "test": "Edge Bypass", "severity": "HIGH",
                    "detail": f"Bypass: path={technique['path']} "
                              f"headers={technique['headers']}",
                })

    def print_report(self):
        """แสดงรายงาน"""
        print(f"\n{'='*50}")
        print(f"Edge Security Test Report")
        print(f"Target: {self.base_url}")
        print(f"{'='*50}")

        high = [r for r in self.results if r["severity"] == "HIGH"]
        med = [r for r in self.results if r["severity"] == "MEDIUM"]

        print(f"\nFindings: {len(self.results)} "
              f"(High: {len(high)}, Medium: {len(med)})")

        for r in self.results:
            print(f"\n  [{r['severity']}] {r['test']}")
            print(f"    {r['detail']}")

# ตัวอย่าง
# tester = EdgeSecurityTester("https://edge.example.com", "token123")
# tester.test_cors()
# tester.test_cache_poisoning()
# tester.test_rate_limiting()
# tester.test_edge_bypass()
# tester.print_report()

Edge Security Checklist

• HTTPS Everywhere: บังคับ HTTPS ทุก Edge Location ใช้ HSTS Header
• Authentication: ใช้ JWT หรือ mTLS สำหรับ Edge-to-Origin Authentication ตรวจสอบ Token ทุก Edge Node
• Cache Control: ไม่ Cache Response ที่มีข้อมูลส่วนตัว ใช้ Cache-Control: private สำหรับ User-specific Data
• Rate Limiting: ตั้ง Rate Limit ทั้งที่ Edge และ Origin ป้องกัน DDoS และ Brute Force
• WAF Rules: ตั้ง WAF Rules ที่ Edge ป้องกัน OWASP Top 10
• Input Validation: Validate Input ทั้งที่ Edge และ Origin อย่าพึ่ง Edge เพียงอย่างเดียว
• Logging: เก็บ Log จากทุก Edge Location รวมศูนย์สำหรับ SIEM Analysis

Burp Suite Pro คืออะไร

เครื่องมือทดสอบความปลอดภัยเว็บแอปพลิเคชันชั้นนำ มี Proxy, Scanner, Intruder, Repeater ใช้หาช่องโหว่ SQL Injection, XSS, CSRF, Authentication Bypass และ API Vulnerabilities

สรุป

Burp Suite Pro เป็นเครื่องมือสำคัญสำหรับทดสอบ Security ของ Edge Computing Applications ต้องตั้งค่า Scope ให้ครอบคลุมทุก Edge Endpoints ใช้ Custom Extensions สำหรับ Edge-specific Vulnerabilities เช่น Cache Poisoning, CORS Misconfiguration, Edge Bypass ทำ Automated Testing ด้วย Script และตรวจสอบตาม Security Checklist อย่างสม่ำเสมอ