安全编排自动化与响应平台安全取证深度分析

XSOAR_HOST="https://xsoar.example.com"
XSOAR_KEY="your-api-key"
PLAYBOOK_ID="pb_12345"

curl -s -H "x-xsoar-key: ${XSOAR_KEY}" \
  "${XSOAR_HOST}/api/v1/playbooks/${PLAYBOOK_ID}" | \
  python3 -m json.tool > current_playbook.json

curl -s -H "x-xsoar-key: ${XSOAR_KEY}" \
  "${XSOAR_HOST}/api/v1/playbooks/${PLAYBOOK_ID}/revisions" | \
  python3 -m json.tool > playbook_history.json

sha256sum current_playbook.json playbook_history.json

import json
import re
import sys
from datetime import datetime

SUSPICIOUS_PATTERNS = [
    r';\s*(rm|cat|curl|wget|python|bash|sh)\s',
    r'\|\s*(base64|nc|netcat|socat)\s',
    r'>\s*/etc/',
    r'\.\./\.\./',
    r'UNION\s+SELECT',
    r'eval\s*\(',
    r'exec\s*\(',
    r'__import__',
    r'\$\{.*\}',
    r'\{\{.*\}\}',
]

def analyze_playbook_inputs(log_path):
    findings = []
    with open(log_path, 'r') as f:
        entries = json.load(f)

    for entry in entries:
        inputs = entry.get('input_parameters', {})
        for param, value in inputs.items():
            if not isinstance(value, str):
                continue
            for pattern in SUSPICIOUS_PATTERNS:
                if re.search(pattern, value, re.IGNORECASE):
                    findings.append({
                        'timestamp': entry.get('timestamp'),
                        'playbook_id': entry.get('playbook_id'),
                        'parameter': param,
                        'value': value[:200],
                        'pattern_matched': pattern,
                        'severity': 'CRITICAL'
                    })

    return findings

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <playbook_execution_log.json>")
        sys.exit(1)

    results = analyze_playbook_inputs(sys.argv[1])
    print(json.dumps(results, indent=2, ensure_ascii=False))

#!/bin/bash

LOG_FILE="${1:-/var/log/xsoar/api_access.log}"
SUSPICIOUS_THRESHOLD=100
TIME_WINDOW=60

echo "SOAR API Token Usage Anomaly Analysis"
echo "======================================"
echo ""

echo "[*] Analyzing unique tokens per source IP..."
awk -F'|' '{print $1, $4}' "$LOG_FILE" | \
  sort | uniq -c | sort -rn | head -20

echo ""
echo "[*] Detecting tokens used from multiple IPs..."
awk -F'|' '{print $4}' "$LOG_FILE" | \
  sort | uniq -c | sort -rn | \
  awk '$1 > 1 {print "[!] Token", $2, "used from", $1, "different IPs"}'

echo ""
echo "[*] Detecting high-frequency token usage (> ${SUSPICIOUS_THRESHOLD}/min)..."
awk -F'|' -v threshold="$SUSPICIOUS_THRESHOLD" '
{
  split($3, t, " ");
  key = $4 " " t[1]
  count[key]++
}
END {
  for (k in count) {
    if (count[k] > threshold) {
      print "[!] " k " => " count[k] " requests"
    }
  }
}' "$LOG_FILE" | sort -t= -k2 -rn

echo ""
echo "[*] Detecting off-hours API access (22:00-06:00)..."
awk -F'|' '{
  split($3, t, ":");
  hour = t[1] + 0;
  if (hour >= 22 || hour < 6) {
    print "[!] Off-hours access:", $0
  }
}' "$LOG_FILE" | head -20

SIEM_LOG="/var/log/siem/security.log"
SOAR_LOG="/var/log/soar/incoming_alerts.log"

echo "SIEM-SOAR Log Integrity Verification"
echo "======================================"

echo "[*] Counting events in SIEM log..."
SIEM_COUNT=$(wc -l < "$SIEM_LOG")
echo "    SIEM event count: ${SIEM_COUNT}"

echo "[*] Counting events in SOAR log..."
SOAR_COUNT=$(wc -l < "$SOAR_LOG")
echo "    SOAR event count: ${SOAR_COUNT}"

echo "[*] Checking event ID gaps..."
awk -F'|' '{print $3}' "$SIEM_LOG" | sort -n > /tmp/siem_ids.txt
awk -F'|' '{print $3}' "$SOAR_LOG" | sort -n > /tmp/soar_ids.txt

diff /tmp/siem_ids.txt /tmp/soar_ids.txt | grep '<' | head -20

echo "[*] Checking chronological order..."
awk -F'|' '{print $2}' "$SOAR_LOG" | awk '{
  if (NR > 1 && $1 < prev) {
    print "[!] Out of order at line " NR ": " prev " > " $1
  }
  prev = $1
}' | head -10

echo "[*] Checking for duplicate event IDs..."
awk -F'|' '{print $3}' "$SOAR_LOG" | sort | uniq -d | head -10

echo "[*] Computing log file hashes for baseline..."
sha256sum "$SIEM_LOG" "$SOAR_LOG"

SOAR_API="https://soar.example.com/api/v1"
SOAR_KEY="your-api-key"
HEARTBEAT_INTERVAL=300
MAX_MISSING_ALERTS=50

echo "SIEM-SOAR Link Health Monitor"
echo "=============================="

LAST_ALERT=$(curl -s -H "x-api-key: ${SOAR_KEY}" \
  "${SOAR_API}/alerts?sort=-timestamp&limit=1" | \
  python3 -c "import json,sys; d=json.load(sys.stdin); print(d[0]['timestamp'])" 2>/dev/null)

echo "[*] Last alert received: ${LAST_ALERT}"

LAST_EPOCH=$(date -d "${LAST_ALERT}" +%s 2>/dev/null || date -j -f "%Y-%m-%dT%H:%M:%S" "${LAST_ALERT}" +%s 2>/dev/null)
NOW_EPOCH=$(date +%s)
GAP_SECONDS=$((NOW_EPOCH - LAST_EPOCH))

echo "[*] Gap since last alert: ${GAP_SECONDS} seconds"

if [ "$GAP_SECONDS" -gt "$((HEARTBEAT_INTERVAL * 3))" ]; then
    echo "[!] WARNING: SIEM-SOAR link may be down!"
    echo "[*] Initiating link recovery procedure..."
    echo "[!] Checking SOAR service status..."
    curl -s -o /dev/null -w "%{http_code}" "${SOAR_API}/health"
    echo ""
fi

SOAR_PLAYBOOK_DIR="/opt/soar/playbooks"
SOAR_CONFIG_DIR="/opt/soar/config"

echo "SOAR Connector Credential Audit"
echo "================================"

echo "[*] Scanning for hardcoded credentials in Playbooks..."
grep -rn -iE '(password|secret|api_key|token|credential)\s*=\s*["\x27]' \
  "$SOAR_PLAYBOOK_DIR" 2>/dev/null | head -30

echo ""
echo "[*] Scanning config files for plaintext secrets..."
find "$SOAR_CONFIG_DIR" -type f \( -name "*.json" -o -name "*.yaml" -o -name "*.yml" -o -name "*.conf" \) \
  -exec grep -l -iE '(password|secret|api_key|token)' {} \; 2>/dev/null

echo ""
echo "[*] Checking for weak credential patterns..."
grep -rn -E '(admin|root|password|123456|default)' \
  "$SOAR_CONFIG_DIR" 2>/dev/null | head -20

echo ""
echo "[*] Listing files with world-readable permissions..."
find "$SOAR_PLAYBOOK_DIR" "$SOAR_CONFIG_DIR" \
  -type f -perm -o+r -name "*.json" -o -name "*.yaml" 2>/dev/null | head -20

SOAR_HOME="/opt/soar"
CONNECTOR_LOG="${SOAR_HOME}/logs/connectors.log"

echo "SOAR Connector Security Audit"
echo "=============================="

echo "[*] Listing active connectors..."
grep -c "CONNECTOR_START" "$CONNECTOR_LOG"

echo ""
echo "[*] Checking for failed authentication attempts..."
grep "AUTH_FAIL" "$CONNECTOR_LOG" | \
  awk '{print $1, $2, $5}' | sort | uniq -c | sort -rn | head -20

echo ""
echo "[*] Identifying connectors with high error rates..."
awk '/CONNECTOR_CALL/ {
  split($0, a, " ");
  connector = a[5];
  calls[connector]++
}
/AUTH_FAIL|TIMEOUT|ERROR/ {
  split($0, a, " ");
  connector = a[5];
  errors[connector]++
}
END {
  for (c in calls) {
    rate = (errors[c]+0) / calls[c] * 100;
    if (rate > 10) {
      printf "[!] %s: %.1f%% error rate (%d/%d)\n", c, rate, errors[c]+0, calls[c]
    }
  }
}' "$CONNECTOR_LOG"

echo ""
echo "[*] Detecting off-hours connector activity..."
awk '/CONNECTOR_CALL/ {
  split($3, t, ":");
  hour = t[1] + 0;
  if (hour >= 22 || hour < 6) {
    print "[!] Off-hours:", $0
  }
}' "$CONNECTOR_LOG" | head -20

title: SOAR Platform Suspicious API Access Pattern
id: a7c3e8f2-4b1d-4e9a-8f2c-6d5e3a1b7c9d
status: experimental
description: Detects suspicious API access patterns on SOAR platforms including brute force, enumeration, and off-hours access
references:
  - https://attack.mitre.org/techniques/T1078/
  - https://attack.mitre.org/techniques/T1110/
author: x7peeps蓝队
date: 2026-07-15
modified: 2026-07-15
tags:
  - attack.credential_access
  - attack.t1078
  - attack.t1110
  - attack.defense_evasion
  - attack.t1070
logsource:
  category: api_access
  product: soar
detection:
  selection_bruteforce:
    status_code: 401
  selection_enumeration:
    status_code: 404
    uri|contains:
      - '/api/v1/playbooks'
      - '/api/v1/connectors'
      - '/api/v1/incidents'
  selection_offhours:
    timestamp|time: '>=22:00|<=06:00'
  selection_bulk_export:
    uri|contains:
      - '/api/v1/export'
      - '/api/v1/incidents/bulk'
    bytes_out|gt: 10485760
  condition: selection_bruteforce | count() by src_ip > 10 within 5m or selection_enumeration | count() by src_ip > 20 within 1m or selection_offhours and selection_bulk_export
falsepositives:
  - Scheduled automation tasks
  - Legitimate bulk export during incident response
level: high

title: SOAR Playbook Unauthorized Modification
id: b8d4f9e3-5c2e-4f0a-9g3d-7e6f4b2c8d0e
status: experimental
description: Detects unauthorized modifications to SOAR Playbooks including logic changes, action replacements, and new step additions
references:
  - https://attack.mitre.org/techniques/T1565/
  - https://attack.mitre.org/techniques/T1562/
author: x7peeps蓝队
date: 2026-07-15
modified: 2026-07-15
tags:
  - attack.impact
  - attack.t1565
  - attack.t1562
  - attack.defense_evasion
  - attack.t1070
logsource:
  category: playbook_audit
  product: soar
detection:
  selection_modification:
    event_type: 'playbook_modified'
  selection_critical_actions:
    playbook_name|contains:
      - 'incident_response'
      - 'malware_containment'
      - 'network_isolation'
      - 'credential_reset'
  selection_offhours_mod:
    timestamp|time: '>=22:00|<=06:00'
  selection_bulk_changes:
    changes_count|gt: 10
  condition: selection_modification and selection_critical_actions and (selection_offhours_mod or selection_bulk_changes)
falsepositives:
  - Planned Playbook updates during maintenance windows
  - Emergency response modifications during active incidents
level: critical

title: SOAR Connector Anomalous External Communication
id: c9e5a0f4-6d3f-4a1b-a04e-8f7g5c3d9e1f
status: experimental
description: Detects anomalous external communication patterns from SOAR connectors indicating potential data exfiltration or compromised credentials
references:
  - https://attack.mitre.org/techniques/T1041/
  - https://attack.mitre.org/techniques/T1199/
author: x7peeps蓝队
date: 2026-07-15
modified: 2026-07-15
tags:
  - attack.exfiltration
  - attack.t1041
  - attack.command_and_control
  - attack.t1199
logsource:
  category: connector_activity
  product: soar
detection:
  selection_new_endpoint:
    event_type: 'connector_call'
    destination|re: '^(?!.*\.(yourcompany|internal)\.com)'
  selection_high_volume:
    bytes_out|gt: 5242880
  selection_offhours:
    timestamp|time: '>=22:00|<=06:00'
  condition: selection_new_endpoint and selection_high_volume or selection_offhours and selection_new_endpoint | count() by connector_id > 5 within 10m
falsepositives:
  - New legitimate integration deployment
  - Scheduled data sync to partner services
level: high

#!/bin/bash

SOAR_BASE="/opt/soar"
AUDIT_LOG="${SOAR_BASE}/logs/audit.log"
HASH_BASELINE="${SOAR_BASE}/.integrity_baseline.json"
REPORT_FILE="/tmp/soar_integrity_report_$(date +%Y%m%d_%H%M%S).json"

echo "SOAR Platform Log Integrity Verification"
echo "========================================="
echo ""

compute_file_hash() {
    find "$1" -type f -name "*.log" -exec sha256sum {} \; 2>/dev/null | sort
}

load_baseline() {
    if [ -f "$HASH_BASELINE" ]; then
        cat "$HASH_BASELINE"
    else
        echo '{"hashes": {}, "last_check": null}'
    fi
}

save_baseline() {
    echo "$1" > "$HASH_BASELINE"
}

check_log_integrity() {
    local log_dir="$1"
    local label="$2"
    local findings=()

    if [ ! -d "$log_dir" ]; then
        echo "[!] ${label} directory not found: ${log_dir}"
        return
    fi

    echo "[*] Checking ${label} logs..."

    for logfile in "$log_dir"/*.log; do
        [ -f "$logfile" ] || continue
        filename=$(basename "$logfile")
        current_hash=$(sha256sum "$logfile" | awk '{print $1}')

        previous_hash=$(python3 -c "
import json, sys
try:
    data = json.load(open('${HASH_BASELINE}'))
    print(data.get('hashes', {}).get('${filename}', ''))
except:
    print('')
" 2>/dev/null)

        if [ -n "$previous_hash" ] && [ "$previous_hash" != "$current_hash" ]; then
            echo "  [!] TAMPERED: ${filename}"
            echo "      Previous: ${previous_hash:0:16}..."
            echo "      Current:  ${current_hash:0:16}..."
        elif [ -z "$previous_hash" ]; then
            echo "  [?] NEW: ${filename}"
        else
            echo "  [+] OK: ${filename}"
        fi
    done
}

check_log_ordering() {
    local logfile="$1"
    echo ""
    echo "[*] Checking chronological order of ${logfile}..."

    awk -F'|' '{
        if (NR > 1 && $2 < prev_ts) {
            printf "  [!] Out of order at line %d: %s > %s\n", NR, prev_ts, $2
            count++
        }
        prev_ts = $2
    }
    END {
        if (!count) print "  [+] All timestamps in order"
    }' "$logfile"
}

check_disk_usage() {
    echo ""
    echo "[*] Checking SOAR log disk usage..."
    du -sh "$SOAR_BASE/logs/" 2>/dev/null

    echo ""
    echo "[*] Checking for unusually large log files..."
    find "$SOAR_BASE/logs/" -name "*.log" -size +100M -exec \
        ls -lh {} \; 2>/dev/null | awk '{print "  [!] Large file:", $NF, $5}'
}

check_retention_policy() {
    echo ""
    echo "[*] Checking log retention (files older than 90 days)..."
    find "$SOAR_BASE/logs/" -name "*.log" -mtime +90 -exec \
        ls -lh {} \; 2>/dev/null | awk '{print "  [?] Old file:", $NF, $5}'
}

main() {
    echo "[*] Audit log: ${AUDIT_LOG}"
    echo "[*] Baseline: ${HASH_BASELINE}"
    echo "[*] Report: ${REPORT_FILE}"
    echo ""

    check_log_integrity "${SOAR_BASE}/logs/audit" "Audit"
    check_log_integrity "${SOAR_BASE}/logs/connectors" "Connector"
    check_log_integrity "${SOAR_BASE}/logs/playbooks" "Playbook"
    check_log_integrity "${SOAR_BASE}/logs/api" "API"

    check_log_ordering "$AUDIT_LOG"
    check_disk_usage
    check_retention_policy

    echo ""
    echo "[*] Generating integrity report..."
    timestamp=$(date -u +%Y-%m-%dT%H:%M:%SZ)
    python3 -c "
import json, hashlib, os
report = {
    'timestamp': '${timestamp}',
    'auditor': 'soar_integrity_checker',
    'baseline_file': '${HASH_BASELINE}',
    'status': 'COMPLETED'
}
with open('${REPORT_FILE}', 'w') as f:
    json.dump(report, f, indent=2)
print('[+] Report saved to ${REPORT_FILE}')
"
}

main

import json
import hashlib
import sys
import os
from datetime import datetime, timedelta
from collections import defaultdict

class PlaybookIntegrityChecker:
    def __init__(self, soar_api_url, api_key):
        self.api_url = soar_api_url.rstrip('/')
        self.api_key = api_key
        self.findings = []
        self.baseline_file = os.path.expanduser('~/.soar_playbook_baseline.json')

    def load_baseline(self):
        if os.path.exists(self.baseline_file):
            with open(self.baseline_file, 'r') as f:
                return json.load(f)
        return {'playbooks': {}, 'last_check': None}

    def save_baseline(self, baseline):
        baseline['last_check'] = datetime.now().isoformat()
        with open(self.baseline_file, 'w') as f:
            json.dump(baseline, f, indent=2)

    def compute_playbook_hash(self, playbook_content):
        canonical = json.dumps(playbook_content, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(canonical.encode('utf-8')).hexdigest()

    def detect_critical_step_changes(self, old_pb, new_pb):
        critical_keywords = [
            'isolate', 'quarantine', 'block', 'shutdown',
            'release', 'disable', 'delete', 'exfiltrate',
            'export', 'forward', 'allow', 'permit'
        ]
        changes = []
        old_steps = old_pb.get('tasks', [])
        new_steps = new_pb.get('tasks', [])

        old_names = {s.get('name', ''): s for s in old_steps}
        new_names = {s.get('name', ''): s for s in new_steps}

        removed = set(old_names.keys()) - set(new_names.keys())
        added = set(new_names.keys()) - set(old_names.keys())

        for step_name in removed:
            if any(kw in step_name.lower() for kw in critical_keywords):
                changes.append({
                    'type': 'CRITICAL_STEP_REMOVED',
                    'step': step_name,
                    'severity': 'CRITICAL'
                })

        for step_name in added:
            if any(kw in step_name.lower() for kw in critical_keywords):
                changes.append({
                    'type': 'CRITICAL_STEP_ADDED',
                    'step': step_name,
                    'severity': 'HIGH'
                })

        for step_name in set(old_names.keys()) & set(new_names.keys()):
            old_step = old_names[step_name]
            new_step = new_names[step_name]
            if json.dumps(old_step, sort_keys=True) != json.dumps(new_step, sort_keys=True):
                severity = 'HIGH' if any(kw in step_name.lower() for kw in critical_keywords) else 'MEDIUM'
                changes.append({
                    'type': 'STEP_MODIFIED',
                    'step': step_name,
                    'severity': severity
                })

        return changes

    def detect_condition_tampering(self, old_pb, new_pb):
        tampering = []
        old_conditions = old_pb.get('conditions', [])
        new_conditions = new_pb.get('conditions', [])

        if len(new_conditions) < len(old_conditions):
            tampering.append({
                'type': 'CONDITIONS_REMOVED',
                'removed_count': len(old_conditions) - len(new_conditions),
                'severity': 'HIGH'
            })

        for i, (old_c, new_c) in enumerate(zip(old_conditions, new_conditions)):
            if old_c != new_c:
                tampering.append({
                    'type': 'CONDITION_MODIFIED',
                    'condition_index': i,
                    'old_value': str(old_c)[:100],
                    'new_value': str(new_c)[:100],
                    'severity': 'MEDIUM'
                })

        return tampering

    def analyze_modification_pattern(self, modifications):
        if not modifications:
            return None

        timestamps = [m.get('timestamp') for m in modifications if m.get('timestamp')]
        if not timestamps:
            return None

        off_hours = sum(1 for ts in timestamps
                       if self._is_off_hours(ts))

        return {
            'total_modifications': len(modifications),
            'off_hours_modifications': off_hours,
            'off_hours_ratio': off_hours / len(timestamps) if modifications else 0,
            'modification_frequency': self._calculate_frequency(timestamps),
            'suspicious_pattern': off_hours / max(len(timestamps), 1) > 0.3
        }

    def _is_off_hours(self, timestamp_str):
        try:
            if 'T' in timestamp_str:
                hour = int(timestamp_str.split('T')[1].split(':')[0])
            else:
                hour = int(timestamp_str.split(' ')[1].split(':')[0])
            return hour >= 22 or hour < 6
        except (IndexError, ValueError):
            return False

    def _calculate_frequency(self, timestamps):
        if len(timestamps) < 2:
            return 0
        try:
            parsed = sorted([datetime.fromisoformat(ts.replace('Z', '+00:00'))
                           for ts in timestamps])
            intervals = [(parsed[i+1] - parsed[i]).total_seconds()
                        for i in range(len(parsed)-1)]
            return sum(intervals) / len(intervals) if intervals else 0
        except ValueError:
            return 0

    def check_playbook_integrity(self, playbooks_data, baseline):
        for pb in playbooks_data:
            pb_id = pb.get('id', 'unknown')
            pb_name = pb.get('name', 'unknown')
            current_hash = self.compute_playbook_hash(pb)
            saved_hash = baseline['playbooks'].get(pb_id, {}).get('hash')

            if saved_hash and saved_hash != current_hash:
                old_content = baseline['playbooks'][pb_id].get('content', {})
                critical_changes = self.detect_critical_step_changes(old_content, pb)
                condition_changes = self.detect_condition_tampering(old_content, pb)

                finding = {
                    'playbook_id': pb_id,
                    'playbook_name': pb_name,
                    'type': 'HASH_MISMATCH',
                    'previous_hash': saved_hash[:16],
                    'current_hash': current_hash[:16],
                    'critical_changes': critical_changes,
                    'condition_changes': condition_changes,
                    'timestamp': datetime.now().isoformat()
                }

                if critical_changes:
                    finding['severity'] = 'CRITICAL'
                elif condition_changes:
                    finding['severity'] = 'HIGH'
                else:
                    finding['severity'] = 'MEDIUM'

                self.findings.append(finding)

            baseline['playbooks'][pb_id] = {
                'hash': current_hash,
                'content': pb,
                'last_seen': datetime.now().isoformat()
            }

    def generate_report(self):
        report = {
            'scan_timestamp': datetime.now().isoformat(),
            'total_playbooks_checked': len(self.findings) + len([f for f in self.findings]),
            'findings': self.findings,
            'severity_summary': defaultdict(int),
            'recommendations': []
        }

        for f in self.findings:
            report['severity_summary'][f.get('severity', 'UNKNOWN')] += 1

        if report['severity_summary'].get('CRITICAL', 0) > 0:
            report['recommendations'].append('立即审查被篡改的 Playbook 并回滚到安全版本')
            report['recommendations'].append('撤销所有近期的 API Token 并强制重新认证')

        if report['severity_summary'].get('HIGH', 0) > 0:
            report['recommendations'].append('审查 Playbook 修改者的身份和权限')
            report['recommendations'].append('检查相关连接器的日志是否存在异常')

        report['recommendations'].append('启用 Playbook 变更的多因素审批流程')

        return report

def main():
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <xsoar_url> <api_key> [playbooks_json]")
        print(f"  playbooks_json: Optional file with pre-exported Playbook data")
        sys.exit(1)

    api_url = sys.argv[1]
    api_key = sys.argv[2]

    checker = PlaybookIntegrityChecker(api_url, api_key)
    baseline = checker.load_baseline()

    if len(sys.argv) > 3:
        with open(sys.argv[3], 'r') as f:
            playbooks = json.load(f)
    else:
        playbooks = []

    checker.check_playbook_integrity(playbooks, baseline)
    checker.save_baseline(baseline)

    report = checker.generate_report()
    print(json.dumps(report, indent=2, ensure_ascii=False, default=str))

if __name__ == '__main__':
    main()

python3 << 'PYTHON'
import base64

def decode_sunburst_subdomain(subdomain):
    encoded_part = subdomain.split('.')[0]
    padded = encoded_part + '=' * (8 - len(encoded_part) % 8)
    try:
        decoded = base64.b32decode(padded)
        return decoded.hex()
    except Exception:
        return "DECODE_FAILED"

suspicious_domains = [
    "dnsyncUpdate.avsvmcloud.com",
    "freescanonline.com",
    "deftsecurity.com",
]

for domain in suspicious_domains:
    parts = domain.split('.')
    if len(parts) > 2:
        sub = parts[0]
        decoded = decode_sunburst_subdomain(sub)
        print(f"[INFO] {domain} => decoded: {decoded}")
PYTHON

SOAR_LOG="/var/log/soar/response_actions.log"

echo "JBS-Style Attack Detection in SOAR Logs"
echo "========================================="

echo "[*] Checking for EDR Agent heartbeat anomalies..."
grep -i "edr.*heartbeat" "$SOAR_LOG" | \
  awk '{print $1, $2, $4, $5}' | \
  sort | uniq -c | sort -rn | head -20

echo ""
echo "[*] Checking for firewall rule modifications..."
grep -i "firewall.*rule.*add" "$SOAR_LOG" | \
  grep -v "approved_change" | head -10

echo ""
echo "[*] Checking for EDR disable events followed by heartbeat..."
grep -n "edr.*disable\|edr.*stop" "$SOAR_LOG" | while read line; do
  linenum=$(echo "$line" | cut -d: -f1)
  nextlines=$(sed -n "$((linenum+1)),$((linenum+5))p" "$SOAR_LOG")
  if echo "$nextlines" | grep -q "edr.*heartbeat"; then
    echo "[!] SUSPICIOUS: EDR disable followed by fake heartbeat at line $linenum"
    echo "$line"
  fi
done