title: DNS Rebinding Suspicious Resolution Patternid: 7e8f9a0b-1c2d-3e4f-5a6b-7c8d9e0f1a2bstatus: experimentaldescription: Detects DNS resolution patterns consistent with DNS rebinding attacks targeting CDN-protected assetsauthor: x7peeps蓝队date: 2026-07-17tags:
- attack.defense_evasion - attack.t1568 - attack.t1568.002logsource:
category: dnsproduct: dns_serverdetection:
selection_low_ttl:
dns.response.ttl|lt: 30selection_high_frequency:
dns.query|contains: '.'selection_private_ip_response:
dns.response|re: '^(10\.|172\.(1[6-9]|2[0-9]|3[01])\.|192\.168\.)'condition: selection_low_ttl and selection_high_frequency and selection_private_ip_responselevel: highfalsepositives:
- Legitimate dynamic DNS services - Internal development environmentsfields:
- dns.query.name - dns.response.address - dns.response.ttl - source.ip
Bash自动化检测脚本
#!/bin/bash
ZONE_ID="${1:-}"API_TOKEN="${2:-}"REPORT_DIR="/tmp/cdn-audit-$(date +%Y%m%d-%H%M%S)"mkdir -p "$REPORT_DIR"echo "=========================================="echo " CDN Security Audit Script v1.0"echo " Target Zone: $ZONE_ID"echo "=========================================="echo ""echo "[*] Step 1: Checking CDN configuration baseline..."curl -s -X GET "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/settings"\
-H "Authorization: Bearer $API_TOKEN"\
-H "Content-Type: application/json" > "$REPORT_DIR/settings.json" 2>/dev/null
python3 -c "
import json
with open('$REPORT_DIR/settings.json') as f:
settings = json.load(f)
for result in settings.get('result', []):
name = result.get('id', 'unknown')
value = result.get('value', 'unknown')
if name in ['min_tls_version', 'tls_1_3', 'ssl', 'security_level']:
print(f'[INFO] {name}: {value}')
if name == 'min_tls_version' and value in ['1.0', '1.1']:
print(f'[!] CRITICAL: {name} is set to {value} (should be 1.2+)')
if name == 'security_level' and value == 'off':
print(f'[!] CRITICAL: Security level is OFF')
"echo ""echo "[*] Step 2: Auditing WAF rules..."curl -s -X GET "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/firewall/rules?per_page=50"\
-H "Authorization: Bearer $API_TOKEN"\
-H "Content-Type: application/json" > "$REPORT_DIR/firewall_rules.json" 2>/dev/null
python3 -c "
import json
with open('$REPORT_DIR/firewall_rules.json') as f:
rules = json.load(f)
total = len(rules.get('result', []))
paused = [r for r in rules.get('result', []) if r.get('paused', False)]
blocked = [r for r in rules.get('result', []) if r.get('action') == 'block']
print(f'[INFO] Total rules: {total}')
print(f'[INFO] Blocked rules: {len(blocked)}')
if paused:
print(f'[!] WARNING: {len(paused)} rules are paused:')
for r in paused:
print(f' - {r.get(\"id\", \"N/A\")} | Description: {r.get(\"description\", \"N/A\")}')
"echo ""echo "[*] Step 3: Checking for unauthenticated Worker endpoints..."curl -s -X GET "https://api.cloudflare.com/client/v4/accounts/$ZONE_ID/workers/scripts"\
-H "Authorization: Bearer $API_TOKEN"\
-H "Content-Type: application/json" > "$REPORT_DIR/workers.json" 2>/dev/null
python3 -c "
import json
with open('$REPORT_DIR/workers.json') as f:
scripts = json.load(f)
for script in scripts.get('result', []):
name = script.get('id', 'unknown')
modified = script.get('modified_on', 'unknown')
size = script.get('size', 0)
print(f'[INFO] Worker: {name} | Modified: {modified} | Size: {size} bytes')
if size > 1048576:
print(f'[!] WARNING: Worker {name} has unusually large size ({size} bytes)')
"echo ""echo "[*] Step 4: Checking DNS configuration..."curl -s -X GET "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/dns_records?per_page=100"\
-H "Authorization: Bearer $API_TOKEN"\
-H "Content-Type: application/json" > "$REPORT_DIR/dns_records.json" 2>/dev/null
python3 -c "
import json
with open('$REPORT_DIR/dns_records.json') as f:
records = json.load(f)
for r in records.get('result', []):
rtype = r.get('type', '')
name = r.get('name', '')
content = r.get('content', '')
ttl = r.get('ttl', 0)
if ttl == 1 and rtype in ['A', 'AAAA', 'CNAME']:
print(f'[!] WARNING: {name} ({rtype}) has TTL=1 (Auto) - potential rebinding risk')
if rtype == 'A' and any(content.startswith(p) for p in ['10.', '172.16.', '192.168.']):
print(f'[!] WARNING: {name} resolves to private IP: {content}')
"echo ""echo "[*] Step 5: Checking SSL/TLS certificate status..."curl -s -X GET "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/ssl/verification"\
-H "Authorization: Bearer $API_TOKEN"\
-H "Content-Type: application/json" > "$REPORT_DIR/ssl_status.json" 2>/dev/null
echo ""echo "[*] Step 6: Checking recent audit log for suspicious activities..."curl -s -X GET "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/audit_logs?page=1&per_page=50"\
-H "Authorization: Bearer $API_TOKEN"\
-H "Content-Type: application/json" > "$REPORT_DIR/audit_logs.json" 2>/dev/null
python3 -c "
import json
from datetime import datetime, timedelta
with open('$REPORT_DIR/audit_logs.json') as f:
logs = json.load(f)
suspicious_actions = ['firewall.delete', 'firewall.update', 'waf.downgrade', 'settings.update']
for log in logs.get('result', []):
action = log.get('action', {}).get('type', '')
actor = log.get('actor', {}).get('email', 'unknown')
when = log.get('when', '')
for sa in suspicious_actions:
if sa in action.lower():
print(f'[!] SUSPICIOUS: {actor} performed {action} at {when}')
"echo ""echo "[*] Step 7: Checking for origin IP exposure..."ORIGIN_IP=$(dig +short $(curl -s "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/dns_records?type=A&per_page=1"\
-H "Authorization: Bearer $API_TOKEN" | python3 -c "import sys,json; print(json.load(sys.stdin)['result'][0]['name'])" 2>/dev/null) | head -1)echo "[INFO] CDN edge IP: $ORIGIN_IP"if echo "$ORIGIN_IP" | grep -qE "^(10\.|172\.(1[6-9]|2[0-9]|3[01])\.|192\.168\.)"; then echo "[!] CRITICAL: Resolved IP is a private address - DNS rebinding possible"fiecho ""echo "=========================================="echo "[+] Audit complete. Reports saved to: $REPORT_DIR"echo "=========================================="
Python自动化检测脚本
#!/usr/bin/env python3import requests
import json
import sys
import hashlib
from datetime import datetime, timedelta
from urllib.parse import urlparse
classCDNAuditEngine:
def__init__(self, zone_id, api_token):
self.zone_id = zone_id
self.api_token = api_token
self.base_url ="https://api.cloudflare.com/client/v4" self.headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json" }
self.findings = []
def_get(self, endpoint):
try:
resp = requests.get(f"{self.base_url}{endpoint}", headers=self.headers, timeout=30)
resp.raise_for_status()
return resp.json()
exceptExceptionas e:
return {"success": False, "error": str(e)}
defaudit_ssl_configuration(self):
data = self._get(f"/zones/{self.zone_id}/settings/min_tls_version")
if data.get("success"):
value = data["result"]["value"]
if value in ["1.0", "1.1"]:
self.findings.append({
"category": "TLS Configuration",
"severity": "CRITICAL",
"finding": f"Minimum TLS version set to {value}",
"recommendation": "Upgrade to TLS 1.2 or higher",
"reference": "MITRE ATT&CK T1557" })
else:
self.findings.append({
"category": "TLS Configuration",
"severity": "INFO",
"finding": f"Minimum TLS version: {value}",
"recommendation": "Configuration is acceptable" })
defaudit_waf_rules(self):
data = self._get(f"/zones/{self.zone_id}/firewall/rules?per_page=100")
if data.get("success"):
rules = data.get("result", [])
paused = [r for r in rules if r.get("paused")]
if paused:
self.findings.append({
"category": "WAF Configuration",
"severity": "HIGH",
"finding": f"{len(paused)} WAF rules are paused",
"details": [r.get("description", r.get("id")) for r in paused[:5]],
"recommendation": "Review and unpause WAF rules unless justified",
"reference": "MITRE ATT&CK T1562.001" })
blocked = [r for r in rules if r.get("action") =="block"]
challenge = [r for r in rules if r.get("action") =="challenge"]
self.findings.append({
"category": "WAF Statistics",
"severity": "INFO",
"finding": f"Total: {len(rules)}, Block: {len(blocked)}, Challenge: {len(challenge)}, Paused: {len(paused)}" })
defaudit_dns_records(self):
data = self._get(f"/zones/{self.zone_id}/dns_records?per_page=100")
if data.get("success"):
records = data.get("result", [])
private_ip_records = []
low_ttl_records = []
for r in records:
if r.get("type") in ["A", "AAAA"]:
content = r.get("content", "")
if any(content.startswith(p) for p in ["10.", "172.16.", "192.168."]):
private_ip_records.append(r)
if r.get("ttl") ==1and r.get("type") in ["A", "AAAA", "CNAME"]:
low_ttl_records.append(r)
if private_ip_records:
self.findings.append({
"category": "DNS Security",
"severity": "HIGH",
"finding": f"{len(private_ip_records)} DNS records resolve to private IPs",
"details": [r["name"] for r in private_ip_records[:5]],
"recommendation": "Verify these records are intentional",
"reference": "MITRE ATT&CK T1568" })
if low_ttl_records:
self.findings.append({
"category": "DNS Security",
"severity": "MEDIUM",
"finding": f"{len(low_ttl_records)} DNS records have auto TTL (potential rebinding risk)",
"details": [r["name"] for r in low_ttl_records[:5]],
"recommendation": "Set explicit TTL values for DNS records",
"reference": "MITRE ATT&CK T1568.002" })
defaudit_workers(self):
data = self._get(f"/accounts/{self.zone_id}/workers/scripts")
if data.get("success"):
scripts = data.get("result", [])
for script in scripts:
size = script.get("size", 0)
if size >1048576:
self.findings.append({
"category": "Edge Functions",
"severity": "MEDIUM",
"finding": f"Worker '{script.get('id')}' has unusually large size ({size} bytes)",
"recommendation": "Review worker code for potential malicious payload",
"reference": "MITRE ATT&CK T1195.002" })
defaudit_audit_logs(self):
data = self._get(f"/zones/{self.zone_id}/audit_logs?page=1&per_page=50")
if data.get("success"):
logs = data.get("result", [])
critical_actions = ["firewall.delete", "waf.downgrade", "settings.update", "access.policy.delete"]
for log in logs:
action_type = log.get("action", {}).get("type", "")
actor = log.get("actor", {}).get("email", "unknown")
when = log.get("when", "")
for ca in critical_actions:
if ca in action_type.lower():
self.findings.append({
"category": "Audit Log",
"severity": "CRITICAL",
"finding": f"Suspicious action: {actor} performed {action_type} at {when}",
"recommendation": "Investigate this action immediately",
"reference": "MITRE ATT&CK T1562.001" })
defrun_full_audit(self):
print(f"[*] Starting CDN security audit for zone: {self.zone_id}")
print(f"[*] Timestamp: {datetime.utcnow().isoformat()}Z")
print()
print("[*] Auditing SSL/TLS configuration...")
self.audit_ssl_configuration()
print("[*] Auditing WAF rules...")
self.audit_waf_rules()
print("[*] Auditing DNS records...")
self.audit_dns_records()
print("[*] Auditing Workers/Edge Functions...")
self.audit_workers()
print("[*] Auditing recent audit logs...")
self.audit_audit_logs()
critical = [f for f in self.findings if f["severity"] =="CRITICAL"]
high = [f for f in self.findings if f["severity"] =="HIGH"]
medium = [f for f in self.findings if f["severity"] =="MEDIUM"]
info = [f for f in self.findings if f["severity"] =="INFO"]
print()
print("="*50)
print(" CDN Security Audit Report")
print("="*50)
print(f" CRITICAL: {len(critical)}")
print(f" HIGH: {len(high)}")
print(f" MEDIUM: {len(medium)}")
print(f" INFO: {len(info)}")
print("="*50)
for f in critical:
print(f"\n[!] CRITICAL: {f['finding']}")
print(f" Category: {f['category']}")
print(f" Recommendation: {f['recommendation']}")
for f in high:
print(f"\n[!] HIGH: {f['finding']}")
print(f" Category: {f['category']}")
print(f" Recommendation: {f['recommendation']}")
return self.findings
if __name__ =="__main__":
if len(sys.argv) <3:
print(f"Usage: {sys.argv[0]} <zone_id> <api_token>")
sys.exit(1)
auditor = CDNAuditEngine(sys.argv[1], sys.argv[2])
findings = auditor.run_full_audit()
with open("/tmp/cdn-audit-results.json", "w") as f:
json.dump(findings, f, indent=2, default=str)
print(f"\n[+] Full results saved to /tmp/cdn-audit-results.json")