数字孪生与工业元宇宙安全取证深度分析

nmap -sV -p 4840,4841,4842 --script opcua-info 192.168.1.0/24

python3 -c "
from opcua import Client
import sys

target = 'opc.tcp://192.168.1.100:4840'
try:
    client = Client(target)
    client.connect()
    print('[+] Connected to: ' + target)
    print('[+] Server Certificate: ' + str(client.get_server_certificate()))
    print('[+] Server URI: ' + str(client.get_server_node().get_child(['2:ServerStatus']).get_value()))
    nodes = client.get_objects_node().get_children()
    for node in nodes:
        print('[+] Object: ' + str(node.get_browse_name()))
    client.disconnect()
except Exception as e:
    print('[-] Connection failed: ' + str(e))
"

python3 -c "
from opcua import Client
import sys

target = 'opc.tcp://192.168.1.100:4840'
try:
    client = Client(target)
    client.connect()
    security_mode = client.security_mode
    security_policy = client.security_policy
    print('[*] Security Mode: ' + str(security_mode))
    print('[*] Security Policy: ' + str(security_policy))
    if 'None' in str(security_policy):
        print('[!] CRITICAL: Server accepts None security policy!')
    if security_mode == 1:
        print('[!] WARNING: Sign-only mode, no encryption!')
    client.disconnect()
except Exception as e:
    print('[-] Audit failed: ' + str(e))
"

curl -s "http://192.168.1.100:5000/MTConnectDemo/probe" | xmllint --format -

curl -s "http://192.168.1.100:5000/MTConnectDemo/current?path=//Devices//Component" | xmllint --format -

curl -s "http://192.168.1.100:5000/MTConnectDemo/sample?from=0&count=1000" | xmllint --format -

import json
from datetime import datetime
from collections import defaultdict

log_file = '/var/log/opcua/audit.log'
baseline = {
    'read_requests_per_min': 120,
    'write_requests_per_min': 15,
    'browse_requests_per_min': 30,
    'max_concurrent_sessions': 20
}

alerts = []

with open(log_file, 'r') as f:
    for line in f:
        entry = json.loads(line.strip())
        timestamp = datetime.fromisoformat(entry['timestamp'])
        msg_type = entry['message_type']

        if msg_type == 'Write' and entry.get('node_id', '').startswith('ns=2;s=PLC'):
            alerts.append({
                'level': 'HIGH',
                'time': str(timestamp),
                'source': entry['source_uri'],
                'detail': 'Direct PLC write via OPC UA: ' + entry['node_id'],
                'mitre': 'T1565.001'
            })

        if entry.get('security_policy') == 'None':
            alerts.append({
                'level': 'CRITICAL',
                'time': str(timestamp),
                'source': entry['source_uri'],
                'detail': 'Connection with no security policy',
                'mitre': 'T1040'
            })

        if entry.get('auth_result') == 'failed':
            alerts.append({
                'level': 'MEDIUM',
                'time': str(timestamp),
                'source': entry['source_uri'],
                'detail': 'Authentication failure: ' + entry.get('user', 'unknown'),
                'mitre': 'T1110'
            })

for alert in sorted(alerts, key=lambda x: x['time']):
    print('[{0}] {1} | {2} | {3} | MITRE: {4}'.format(
        alert['level'], alert['time'], alert['source'], alert['detail'], alert['mitre']
    ))

nmap -sV -sC -p 1-65535 --open 192.168.10.0/24 -oX /tmp/dt_scan.xml

nmap --script=http-enum,http-methods,http-title -p 80,443,8080,8443,3000,5000 192.168.10.0/24

nmap --script=opcua-info,modbus-discover,siemens-s7-info -p 4840,502,102,1089 192.168.10.0/24

python3 << 'EOF'
import subprocess
import xml.etree.ElementTree as ET
import json

targets = ['192.168.10.0/24']
services = {
    'OPC UA': [4840, 4841, 4842],
    'MQTT': [1883, 8883],
    'AMQP': [5671, 5672],
    'HTTP/HTTPS': [80, 443, 8080, 8443, 3000, 5000],
    'MODBUS': [502],
    'S7COMM': [102],
    'InfluxDB': [8086],
    'Grafana': [3000],
    'Unity WebGL': [8090],
    'NVIDIA Omniverse': [8211, 8212],
    'RTSP': [554, 8554],
    'SSH': [22]
}

attack_surface = []
for target in targets:
    ports = []
    for svc_name, svc_ports in services.items():
        ports.extend(svc_ports)
    port_str = ','.join(map(str, sorted(set(ports))))
    result = subprocess.run(
        ['nmap', '-sV', '-p', port_str, '--open', '-oX', '-', target],
        capture_output=True, text=True
    )
    root = ET.fromstring(result.stdout)
    for host in root.findall('.//host'):
        addr = host.find('address').get('addr')
        for port_elem in host.findall('.//port'):
            port_id = port_elem.get('portid')
            state = port_elem.find('state')
            if state is not None and state.get('state') == 'open':
                service = port_elem.find('service')
                svc_name = service.get('name', 'unknown') if service is not None else 'unknown'
                attack_surface.append({
                    'host': addr,
                    'port': port_id,
                    'service': svc_name,
                    'risk_level': 'HIGH' if int(port_id) in [4840, 502, 102] else 'MEDIUM'
                })

print(json.dumps(attack_surface, indent=2))
print('[*] Total open services: ' + str(len(attack_surface)))
print('[*] High-risk services: ' + str(sum(1 for s in attack_surface if s['risk_level'] == 'HIGH')))
EOF

import hashlib
import json
import os
import sys
from datetime import datetime

model_registry = '/opt/dt-platform/model-registry/'
baseline_file = '/etc/dt-platform/model-baseline.json'

def calculate_model_hash(model_path):
    sha256_hash = hashlib.sha256()
    with open(model_path, 'rb') as f:
        for byte_block in iter(lambda: f.read(4096), b''):
            sha256_hash.update(byte_block)
    return sha256_hash.hexdigest()

def verify_model_integrity():
    with open(baseline_file, 'r') as f:
        baseline = json.load(f)

    alerts = []
    for model_name, expected in baseline.items():
        model_path = os.path.join(model_registry, model_name)
        if not os.path.exists(model_path):
            alerts.append({
                'severity': 'CRITICAL',
                'model': model_name,
                'issue': 'Model file missing from registry',
                'mitre': 'T1565.001'
            })
            continue

        current_hash = calculate_model_hash(model_path)
        if current_hash != expected['sha256']:
            alerts.append({
                'severity': 'CRITICAL',
                'model': model_name,
                'issue': 'Hash mismatch - expected ' + expected['sha256'][:16] + ' got ' + current_hash[:16],
                'mitre': 'T1565.001'
            })

        mtime = os.path.getmtime(model_path)
        mtime_dt = datetime.fromtimestamp(mtime)
        last_authorized = datetime.fromisoformat(expected['last_authorized'])
        if mtime_dt > last_authorized:
            alerts.append({
                'severity': 'HIGH',
                'model': model_name,
                'issue': 'Model modified after last authorized update: ' + str(mtime_dt),
                'mitre': 'T1565.001'
            })

    return alerts

if __name__ == '__main__':
    results = verify_model_integrity()
    if results:
        for r in results:
            print('[{0}] {1}: {2} | MITRE: {3}'.format(
                r['severity'], r['model'], r['issue'], r['mitre']
            ))
        sys.exit(1)
    else:
        print('[+] All models integrity verified')
        sys.exit(0)

import numpy as np
import json
from scipy import stats

def detect_training_data_poisoning(data_log_path):
    with open(data_log_path, 'r') as f:
        records = json.load(f)

    sensor_values = {}
    for record in records:
        sensor_id = record['sensor_id']
        value = record['value']
        if sensor_id not in sensor_values:
            sensor_values[sensor_id] = []
        sensor_values[sensor_id].append(value)

    anomalies = []
    for sensor_id, values in sensor_values.items():
        arr = np.array(values)
        z_scores = np.abs(stats.zscore(arr))
        outlier_ratio = np.sum(z_scores > 3.0) / len(arr)

        if outlier_ratio > 0.05:
            anomalies.append({
                'sensor': sensor_id,
                'outlier_ratio': round(outlier_ratio, 4),
                'mean': round(float(np.mean(arr)), 2),
                'std': round(float(np.std(arr)), 2),
                'min': round(float(np.min(arr)), 2),
                'max': round(float(np.max(arr)), 2),
                'mitre': 'T1565.002'
            })

        Q1 = np.percentile(arr, 25)
        Q3 = np.percentile(arr, 75)
        IQR = Q3 - Q1
        extreme_outliers = np.sum((arr < Q1 - 3 * IQR) | (arr > Q3 + 3 * IQR))
        if extreme_outliers > 0:
            anomalies.append({
                'sensor': sensor_id,
                'extreme_outliers': int(extreme_outliers),
                'iqr_range': round(float(IQR), 2),
                'mitre': 'T1565.002'
            })

    return anomalies

if __name__ == '__main__':
    results = detect_training_data_poisoning('/var/log/dt-platform/training-data.json')
    for r in results:
        print('[ALERT] Sensor: {0} | Issue: {1} | MITRE: {2}'.format(
            r['sensor'], json.dumps({k: v for k, v in r.items() if k not in ['sensor', 'mitre']}), r['mitre']
        ))

python3 << 'PYEOF'
import json
import time
import hashlib
from datetime import datetime

def collect_scada_snapshot(scada_endpoint):
    snapshot = {
        'timestamp': datetime.utcnow().isoformat(),
        'source': 'scada',
        'readings': {}
    }
    import urllib.request
    try:
        req = urllib.request.Request(scada_endpoint + '/api/v1/readings')
        with urllib.request.urlopen(req, timeout=10) as response:
            data = json.loads(response.read())
            snapshot['readings'] = data
            snapshot['hash'] = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
    except Exception as e:
        snapshot['error'] = str(e)
    return snapshot

def collect_dt_snapshot(dt_endpoint):
    snapshot = {
        'timestamp': datetime.utcnow().isoformat(),
        'source': 'digital_twin',
        'readings': {}
    }
    import urllib.request
    try:
        req = urllib.request.Request(dt_endpoint + '/api/v1/state')
        with urllib.request.urlopen(req, timeout=10) as response:
            data = json.loads(response.read())
            snapshot['readings'] = data
            snapshot['hash'] = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
    except Exception as e:
        snapshot['error'] = str(e)
    return snapshot

def compare_snapshots(scada_snap, dt_snap, tolerance=0.02):
    alerts = []
    scada_data = scada_snap.get('readings', {})
    dt_data = dt_snap.get('readings', {})

    scada_time = datetime.fromisoformat(scada_snap['timestamp'])
    dt_time = datetime.fromisoformat(dt_snap['timestamp'])
    time_diff = abs((dt_time - scada_time).total_seconds())

    if time_diff > 30:
        alerts.append({
            'level': 'HIGH',
            'type': 'TIMESTAMP_DRIFT',
            'detail': 'Time difference between SCADA and DT: ' + str(round(time_diff, 1)) + 's',
            'mitre': 'T1499'
        })

    common_keys = set(scada_data.keys()) & set(dt_data.keys())
    missing_in_dt = set(scada_data.keys()) - set(dt_data.keys())
    extra_in_dt = set(dt_data.keys()) - set(scada_data.keys())

    for key in missing_in_dt:
        alerts.append({
            'level': 'HIGH',
            'type': 'MISSING_MAPPING',
            'detail': 'Tag ' + key + ' exists in SCADA but missing in DT',
            'mitre': 'T1565.001'
        })

    for key in extra_in_dt:
        alerts.append({
            'level': 'MEDIUM',
            'type': 'UNAUTHORIZED_TAG',
            'detail': 'Tag ' + key + ' exists in DT but not in SCADA source',
            'mitre': 'T1565.001'
        })

    for key in common_keys:
        scada_val = float(scada_data[key])
        dt_val = float(dt_data[key])
        if scada_val != 0:
            deviation = abs(scada_val - dt_val) / abs(scada_val)
        else:
            deviation = abs(scada_val - dt_val)

        if deviation > tolerance:
            alerts.append({
                'level': 'CRITICAL',
                'type': 'VALUE_DEVIATION',
                'detail': 'Tag ' + key + ': SCADA=' + str(scada_val) + ' DT=' + str(dt_val) + ' deviation=' + str(round(deviation * 100, 2)) + '%',
                'mitre': 'T1565.001'
            })

    return alerts

if __name__ == '__main__':
    scada = collect_scada_snapshot('http://192.168.10.1:8080')
    dt = collect_dt_snapshot('http://192.168.20.1:5000')
    results = compare_snapshots(scada, dt)
    for r in results:
        print('[{0}] {1}: {2} | MITRE: {3}'.format(r['level'], r['type'], r['detail'], r['mitre']))
    if not results:
        print('[+] SCADA-DT mapping integrity verified - no deviations detected')
PYEOF

import json
import hashlib
from collections import defaultdict
from datetime import datetime

def audit_metaverse_access(log_path):
    with open(log_path, 'r') as f:
        logs = [json.loads(line) for line in f.readlines()]

    user_actions = defaultdict(list)
    role_changes = []
    cross_boundary = []
    privilege_escalation = []

    for log in logs:
        user_id = log.get('user_id', 'unknown')
        action = log.get('action', 'unknown')
        timestamp = log.get('timestamp', '')
        source_ip = log.get('source_ip', '')
        target_resource = log.get('target_resource', '')
        user_role = log.get('user_role', '')

        user_actions[user_id].append({
            'action': action,
            'time': timestamp,
            'resource': target_resource,
            'role': user_role
        })

        if action in ['role_update', 'permission_grant', 'admin_enable']:
            role_changes.append(log)

        if action in ['physical_device_control', 'plc_write', 'emergency_stop']:
            cross_boundary.append(log)

    for user_id, actions in user_actions.items():
        roles_used = set(a['role'] for a in actions)
        if len(roles_used) > 1:
            privilege_escalation.append({
                'user': user_id,
                'roles': list(roles_used),
                'action_count': len(actions),
                'mitre': 'T1098'
            })

    high_freq_users = {uid: len(actions) for uid, actions in user_actions.items()
                       if len(actions) > 1000}
    if high_freq_users:
        for uid, count in high_freq_users.items():
            privilege_escalation.append({
                'user': uid,
                'issue': 'Abnormally high action count: ' + str(count),
                'mitre': 'T1078'
            })

    for log in cross_boundary:
        user_id = log.get('user_id', '')
        actions = user_actions.get(user_id, [])
        roles = set(a['role'] for a in actions)
        if not any('operator' in r.lower() or 'admin' in r.lower() for r in roles):
            privilege_escalation.append({
                'user': user_id,
                'issue': 'Non-privileged user performing physical control: ' + log.get('action', ''),
                'mitre': 'T1078'
            })

    return {
        'role_changes': role_changes,
        'cross_boundary_ops': cross_boundary,
        'privilege_issues': privilege_escalation
    }

if __name__ == '__main__':
    results = audit_metaverse_access('/var/log/metaverse/access.log')
    print('=== Role Changes ===')
    for rc in results['role_changes']:
        print('[ROLE] {0} by {1} at {2}'.format(
            rc['action'], rc.get('user_id', '?'), rc.get('timestamp', '?')
        ))
    print('=== Cross-Boundary Operations ===')
    for cb in results['cross_boundary_ops']:
        print('[CROSS] {0} by {1} on {2}'.format(
            cb['action'], cb.get('user_id', '?'), cb.get('target_resource', '?')
        ))
    print('=== Privilege Issues ===')
    for pi in results['privilege_issues']:
        print('[PRIV] {0}: {1}'.format(pi.get('user', '?'), pi.get('issue', pi.get('roles', ''))))

#!/bin/bash

INFLUX_HOST="http://localhost:8086"
INFLUX_DB="digital_twin_telemetry"
HASH_STORE="/var/lib/dt-platform/integrity-hashes/"

check_data_integrity() {
    local measurement=$1
    local start_time=$(date -u -d '5 minutes ago' '+%Y-%m-%dT%H:%M:%SZ')
    local end_time=$(date -u '+%Y-%m-%dT%H:%M:%SZ')

    local query="SELECT * FROM \"${measurement}\" WHERE time >= '${start_time}' AND time < '${end_time}'"
    local result=$(curl -s -G "${INFLUX_HOST}/query" \
        --data-urlencode "db=${INFLUX_DB}" \
        --data-urlencode "q=${query}" \
        -H "Accept: application/csv")

    local record_count=$(echo "$result" | grep -v "^," | grep -v "^name" | grep -c ",")

    local data_hash=$(echo "$result" | sha256sum | cut -d' ' -f1)
    local hash_file="${HASH_STORE}${measurement}_latest.hash"

    if [ -f "$hash_file" ]; then
        local stored_hash=$(cat "$hash_file")
        if [ "$data_hash" != "$stored_hash" ]; then
            echo "[CRITICAL] Data integrity violation in measurement: ${measurement}"
            echo "  Expected hash: ${stored_hash}"
            echo "  Actual hash:   ${data_hash}"
            echo "  Records in batch: ${record_count}"
            echo "  MITRE: T1565.001"
        else
            echo "[OK] ${measurement}: ${record_count} records verified"
        fi
    else
        echo "[INFO] ${measurement}: First run, storing baseline hash"
    fi

    echo "$data_hash" > "$hash_file"
}

check_data_anomaly() {
    local measurement=$1
    local query="SELECT mean(\"value\"), stddev(\"value\"), count(\"value\") FROM \"${measurement}\" WHERE time >= now() - 1h GROUP BY time(5m)"
    local result=$(curl -s -G "${INFLUX_HOST}/query" \
        --data-urlencode "db=${INFLUX_DB}" \
        --data-urlencode "q=${query}" \
        -H "Accept: application/csv")

    echo "$result" | awk -F',' 'NR>2 && $3 != "" && $4 != "" {
        if ($4 > 0 && ($3 - $2) / $4 > 3) {
            print "[ALERT] Anomaly in " $1 ": mean=" $2 " stddev=" $4 " (>3σ deviation)"
            print "  MITRE: T1499"
        }
    }'
}

echo "=== Digital Twin Data Integrity Check ==="
echo "Timestamp: $(date -u '+%Y-%m-%dT%H:%M:%SZ')"

for measurement in temperature pressure flow_rate vibration power_consumption; do
    check_data_integrity "$measurement"
done

echo ""
echo "=== Statistical Anomaly Detection ==="
for measurement in temperature pressure flow_rate vibration power_consumption; do
    check_data_anomaly "$measurement"
done

echo ""
echo "=== Check Complete ==="

echo "=== Digital Twin Evidence Collection Script ==="
echo "Timestamp: $(date -u '+%Y-%m-%dT%H:%M:%SZ')"
EVIDENCE_DIR="/tmp/dt_evidence_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$EVIDENCE_DIR"

cp -r /var/log/opcua/ "$EVIDENCE_DIR/opcua_logs/" 2>/dev/null
cp -r /var/log/mqtt/ "$EVIDENCE_DIR/mqtt_logs/" 2>/dev/null
cp -r /opt/dt-platform/config/ "$EVIDENCE_DIR/platform_config/" 2>/dev/null
cp -r /opt/dt-platform/model-registry/ "$EVIDENCE_DIR/model_registry/" 2>/dev/null

cp /var/log/metaverse/access.log "$EVIDENCE_DIR/metaverse_access.log" 2>/dev/null
cp /var/log/influxdb/audit.log "$EVIDENCE_DIR/influxdb_audit.log" 2>/dev/null

for model_file in /opt/dt-platform/model-registry/*.pb /opt/dt-platform/model-registry/*.onnx /opt/dt-platform/model-registry/*.h5; do
    if [ -f "$model_file" ]; then
        sha256sum "$model_file" >> "$EVIDENCE_DIR/model_hashes.txt"
    fi
done

echo "Evidence collected to: $EVIDENCE_DIR"
echo "Evidence hash:"
tar czf "/tmp/dt_evidence_$(date +%Y%m%d_%H%M%S).tar.gz" -C "$EVIDENCE_DIR" . | sha256sum
echo "Evidence collection complete"

title: OPC UA Unauthorized PLC Write via Digital Twin Platform
id: 94a0c1e2-5b3f-4d7a-8e1f-2c6d9a0b3e4f
status: stable
description: Detects unauthorized OPC UA write operations targeting PLC registers through the digital twin platform
references:
  - https://www.opcua-products.com/security-best-practices
author: x7peeps Digital Forensics
date: 2026/07/19
tags:
  - attack.initial_access
  - attack.t1190
  - attack.t1565.001
  - ics
  - digital_twin
logsource:
  category: application
  product: opcua
  service: audit
detection:
  selection:
    message_type: Write
    security_policy: None
    node_id|startswith:
      - 'ns=2;s=PLC'
      - 'ns=2;s=Controller'
      - 'ns=2;s=Register'
  filter_main:
    source_uri|contains:
      - 'authorized-gateway'
      - 'maintenance-console'
  condition: selection and not filter_main
falsepositives:
  - Authorized maintenance operations
  - Scheduled PLC configuration updates
level: critical
---
title: Digital Twin Simulation Model Tampering Detection
id: b7e2f4a1-8c3d-4e5f-9a1b-2c3d4e5f6a7b
status: stable
description: Detects unauthorized modification of digital twin simulation model files
references:
  - https://docs.nist.gov/publications/detail/ir/8183/final
author: x7peeps Digital Forensics
date: 2026/07/19
tags:
  - attack.impact
  - attack.t1565.001
  - ics
  - digital_twin
  - simulation
logsource:
  category: file
  product: linux
  service: audit
detection:
  selection:
    type: FILE_MODIFY
    path|contains:
      - '/model-registry/'
      - '/simulation-models/'
      - '/twin-configs/'
    path|endswith:
      - '.pb'
      - '.onnx'
      - '.h5'
      - '.pt'
      - '.json'
      - '.yaml'
  filter_maintenance:
    user: 'dt-updater'
    process_name|contains:
      - 'model-manager'
      - 'hot-reload'
  condition: selection and not filter_maintenance
falsepositives:
  - Scheduled model updates by authorized personnel
  - Automated model training pipeline outputs
level: critical
---
title: Industrial Metaverse Anomalous Session Behavior
id: c8f3d5a2-9e4b-4f6a-8b2c-3d4e5f6a7b8c
status: stable
description: Detects anomalous session behavior in industrial metaverse platforms indicating potential account compromise
references:
  - https://attack.mitre.org/techniques/T1078/
author: x7peeps Digital Forensics
date: 2026/07/19
tags:
  - attack.credential_access
  - attack.t1078
  - attack.t1539
  - industrial_metaverse
logsource:
  category: authentication
  product: metaverse_platform
detection:
  selection_source:
    event_type: 'session_create'
    source_ip|is_not: '192.168.*'
  selection_anomaly:
    event_type: 'session_create'
    concurrent_sessions|gt: 3
    session_duration|gt: 36000
  selection_privilege:
    event_type: 'role_change'
    new_role|contains:
      - 'admin'
      - 'operator'
      - 'emergency_override'
  condition: selection_source or selection_anomaly or selection_privilege
falsepositives:
  - Legitimate remote access by authorized operators
  - Emergency override procedures
level: high

#!/bin/bash

DT_PLATFORM_DIR="/opt/dt-platform"
LOG_DIR="/var/log"
REPORT_DIR="/tmp/dt_hunt_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$REPORT_DIR"

echo "=== Digital Twin Security Hunting Report ==="
echo "Started: $(date -u '+%Y-%m-%dT%H:%M:%SZ')"
echo ""

echo "--- Phase 1: OPC UA Log Analysis ---"
if [ -f "$LOG_DIR/opcua/audit.log" ]; then
    echo "Unauthorized write attempts:" > "$REPORT_DIR/opcua_findings.txt"
    grep -i '"message_type":"Write"' "$LOG_DIR/opcua/audit.log" | \
        grep '"security_policy":"None"' >> "$REPORT_DIR/opcua_findings.txt"
    
    echo "Anonymous connections:" >> "$REPORT_DIR/opcua_findings.txt"
    grep -i '"auth_method":"anonymous"' "$LOG_DIR/opcua/audit.log" | wc -l >> "$REPORT_DIR/opcua_findings.txt"
    
    echo "Failed authentication attempts:" >> "$REPORT_DIR/opcua_findings.txt"
    grep -i '"auth_result":"failed"' "$LOG_DIR/opcua/audit.log" | \
        awk -F'"timestamp":"' '{print $2}' | cut -d'"' -f1 | sort | uniq -c | sort -rn | head -20 >> "$REPORT_DIR/opcua_findings.txt"
    
    echo "[*] OPC UA findings saved to $REPORT_DIR/opcua_findings.txt"
fi

echo ""
echo "--- Phase 2: Simulation Model Integrity Check ---"
MODEL_BASELINE="/etc/dt-platform/model-baseline.json"
if [ -f "$MODEL_BASELINE" ]; then
    echo "Checking model integrity..." > "$REPORT_DIR/model_integrity.txt"
    for model_file in "$DT_PLATFORM_DIR"/model-registry/*.pb "$DT_PLATFORM_DIR"/model-registry/*.onnx; do
        if [ -f "$model_file" ]; then
            current_hash=$(sha256sum "$model_file" | cut -d' ' -f1)
            model_name=$(basename "$model_file")
            expected_hash=$(python3 -c "import json; d=json.load(open('$MODEL_BASELINE')); print(d.get('$model_name',{}).get('sha256','NOT_FOUND'))" 2>/dev/null)
            if [ "$current_hash" != "$expected_hash" ] && [ "$expected_hash" != "NOT_FOUND" ]; then
                echo "[CRITICAL] Model $model_name: HASH MISMATCH" >> "$REPORT_DIR/model_integrity.txt"
                echo "  Expected: $expected_hash" >> "$REPORT_DIR/model_integrity.txt"
                echo "  Actual:   $current_hash" >> "$REPORT_DIR/model_integrity.txt"
            fi
            echo "$model_name:$current_hash" >> "$REPORT_DIR/model_integrity.txt"
        fi
    done
    echo "[*] Model integrity report saved to $REPORT_DIR/model_integrity.txt"
fi

echo ""
echo "--- Phase 3: Network Anomaly Detection ---"
if command -v tcpdump &> /dev/null; then
    echo "Capturing OPC UA traffic snapshot..."
    timeout 30 tcpdump -i any -c 10000 -w "$REPORT_DIR/opcua_capture.pcap" port 4840 2>/dev/null
    
    tshark -r "$REPORT_DIR/opcua_capture.pcap" -T fields \
        -e frame.time -e ip.src -e ip.dst -e tcp.srcport -e tcp.dstport \
        > "$REPORT_DIR/flow_summary.txt" 2>/dev/null
    
    echo "[*] Network capture saved to $REPORT_DIR/flow_summary.txt"
fi

echo ""
echo "--- Phase 4: MQTT Message Anomaly Detection ---"
if [ -f "$LOG_DIR/mqtt/broker.log" ]; then
    echo "MQTT topic distribution:" > "$REPORT_DIR/mqtt_findings.txt"
    grep -oP '"topic":"[^"]*"' "$LOG_DIR/mqtt/broker.log" | sort | uniq -c | sort -rn | head -30 >> "$REPORT_DIR/mqtt_findings.txt"
    
    echo "MQTT unauthorized publish attempts:" >> "$REPORT_DIR/mqtt_findings.txt"
    grep -i '"access":"denied"' "$LOG_DIR/mqtt/broker.log" >> "$REPORT_DIR/mqtt_findings.txt"
    
    echo "[*] MQTT findings saved to $REPORT_DIR/mqtt_findings.txt"
fi

echo ""
echo "--- Phase 5: Metaverse Access Log Analysis ---"
if [ -f "$LOG_DIR/metaverse/access.log" ]; then
    echo "Failed authentication events:" > "$REPORT_DIR/metaverse_findings.txt"
    grep -i '"status":"failed"' "$LOG_DIR/metaverse/access.log" | \
        awk -F'"user_id":"' '{print $2}' | cut -d'"' -f1 | sort | uniq -c | sort -rn | head -20 >> "$REPORT_DIR/metaverse_findings.txt"
    
    echo "Cross-boundary operations:" >> "$REPORT_DIR/metaverse_findings.txt"
    grep -i '"action":"physical_device_control\|plc_write\|emergency_stop"' "$LOG_DIR/metaverse/access.log" >> "$REPORT_DIR/metaverse_findings.txt"
    
    echo "[*] Metaverse findings saved to $REPORT_DIR/metaverse_findings.txt"
fi

echo ""
echo "=== Hunting Report Summary ==="
echo "Report directory: $REPORT_DIR"
echo "Completed: $(date -u '+%Y-%m-%dT%H:%M:%SZ')"
for f in "$REPORT_DIR"/*.txt; do
    if [ -f "$f" ]; then
        echo "  $(basename $f): $(wc -l < $f) lines"
    fi
done

import json
import sys
import numpy as np
from collections import defaultdict
from datetime import datetime, timedelta

def load_sensor_data(data_path):
    with open(data_path, 'r') as f:
        data = json.load(f)
    return data

def detect_temporal_anomalies(data, window_size=60, z_threshold=3.5):
    alerts = []
    readings = data.get('readings', [])

    timestamps = [datetime.fromisoformat(r['timestamp']) for r in readings]
    values = [r['value'] for r in readings]

    for i in range(window_size, len(values)):
        window = values[i - window_size:i]
        current = values[i]

        mean = np.mean(window)
        std = np.std(window)

        if std > 0:
            z_score = abs(current - mean) / std
        else:
            z_score = 0

        if z_score > z_threshold:
            alerts.append({
                'level': 'CRITICAL',
                'type': 'TEMPORAL_ANOMALY',
                'timestamp': str(timestamps[i]),
                'value': current,
                'expected_mean': round(mean, 2),
                'z_score': round(z_score, 2),
                'window_size': window_size,
                'mitre': 'T1565.001'
            })

        if i > 1:
            rate_of_change = abs(values[i] - values[i - 1]) / max(abs(values[i - 1]), 0.001)
            if rate_of_change > 0.5:
                alerts.append({
                    'level': 'HIGH',
                    'type': 'RAPID_CHANGE',
                    'timestamp': str(timestamps[i]),
                    'from_value': values[i - 1],
                    'to_value': values[i],
                    'change_rate': round(rate_of_change * 100, 1),
                    'mitre': 'T1565.001'
                })

    return alerts

def detect_correlation_break(sensor_data_map):
    alerts = []
    sensors = list(sensor_data_map.keys())

    for i in range(len(sensors)):
        for j in range(i + 1, len(sensors)):
            s1 = sensor_data_map[sensors[i]]
            s2 = sensor_data_map[sensors[j]]

            min_len = min(len(s1), len(s2))
            if min_len < 30:
                continue

            arr1 = np.array(s1[-min_len:])
            arr2 = np.array(s2[-min_len:])

            if np.std(arr1) > 0 and np.std(arr2) > 0:
                correlation = np.corrcoef(arr1, arr2)[0, 1]

                if abs(correlation) > 0.8:
                    recent_corr = np.corrcoef(arr1[-30:], arr2[-30:])[0, 1]
                    if abs(recent_corr) < 0.3:
                        alerts.append({
                            'level': 'CRITICAL',
                            'type': 'CORRELATION_BREAK',
                            'sensor_pair': sensors[i] + ' <-> ' + sensors[j],
                            'historical_correlation': round(correlation, 3),
                            'recent_correlation': round(recent_corr, 3),
                            'mitre': 'T1565.001'
                        })

    return alerts

def generate_hunt_report(data_path):
    data = load_sensor_data(data_path)

    all_alerts = []
    sensor_values = defaultdict(list)

    for reading in data.get('readings', []):
        sensor_id = reading.get('sensor_id', 'unknown')
        sensor_values[sensor_id].append(reading.get('value', 0))

    for sensor_id, values in sensor_values.items():
        readings = [r for r in data['readings'] if r.get('sensor_id') == sensor_id]
        sensor_data = {'readings': readings}
        temporal_alerts = detect_temporal_anomalies(sensor_data)
        for alert in temporal_alerts:
            alert['sensor'] = sensor_id
        all_alerts.extend(temporal_alerts)

    correlation_alerts = detect_correlation_break(sensor_values)
    all_alerts.extend(correlation_alerts)

    all_alerts.sort(key=lambda x: x.get('timestamp', ''), reverse=True)

    critical = sum(1 for a in all_alerts if a['level'] == 'CRITICAL')
    high = sum(1 for a in all_alerts if a['level'] == 'HIGH')

    print('=== Digital Twin Security Hunt Report ===')
    print('Data source: ' + data_path)
    print('Sensors analyzed: ' + str(len(sensor_values)))
    print('Total readings: ' + str(sum(len(v) for v in sensor_values.values())))
    print('CRITICAL alerts: ' + str(critical))
    print('HIGH alerts: ' + str(high))
    print('')

    for alert in all_alerts[:50]:
        sensor = alert.get('sensor', 'N/A')
        print('[{0}] {1} | Sensor: {2} | Type: {3} | Detail: {4}'.format(
            alert['level'], alert.get('timestamp', 'N/A'), sensor,
            alert['type'], json.dumps({k: v for k, v in alert.items() if k not in ['level', 'timestamp', 'sensor', 'type']})
        ))

    return all_alerts

if __name__ == '__main__':
    generate_hunt_report('/var/log/dt-platform/sensor-data.json')

rule DigitalTwin_MaliciousModel_Python {
    meta:
        description = "Detects malicious Python scripts targeting digital twin simulation models"
        author = "x7peeps Digital Forensics"
        date = "2026/07/19"
        reference = "Digital Twin Model Tampering"
        mitre_attack = "T1565.001"
        severity = "critical"
    strings:
        $s1 = "model_registry" ascii
        $s2 = "sha256" ascii
        $s3 = "onnx" ascii
        $s4 = "simulink" ascii
        $s5 = "opcua" ascii
        $py1 = "import tensorflow" ascii
        $py2 = "import torch" ascii
        $py3 = "model.load_weights" ascii
        $py4 = "np.save" ascii
        $payload1 = "os.system" ascii
        $payload2 = "subprocess.Popen" ascii
        $payload3 = "eval(" ascii
        $payload4 = "exec(" ascii
        $exfil1 = "requests.post" ascii
        $exfil2 = "socket.socket" ascii
        $exfil3 = "paramiko" ascii
    condition:
        uint32(0) == 0x03f30d0a and
        2 of ($s*) and
        1 of ($py*) and
        1 of ($payload*) and
        1 of ($exfil*)
}

rule DigitalTwin_MaliciousModel_Binary {
    meta:
        description = "Detects tampered ONNX/TensorFlow binary models with embedded payloads"
        author = "x7peeps Digital Forensics"
        date = "2026/07/19"
        reference = "Digital Twin Model Binary Tampering"
        mitre_attack = "T1565.001"
        severity = "critical"
    strings:
        $onnx_magic = { 08 07 0A 0D 0A }
        $tf_magic = { 08 01 12 }
        $py_code = "import os" ascii
        $py_code2 = "subprocess" ascii
        $py_code3 = "eval(" ascii
        $py_code4 = "exec(" ascii
        $net1 = "socket" ascii
        $net2 = "http" ascii
        $net3 = "base64" ascii
    condition:
        ($onnx_magic at 0 or $tf_magic at 0) and
        2 of ($py_code*) and
        1 of ($net*)
}

rule IndustrialMetaverse_PhishingPayload {
    meta:
        description = "Detects phishing payloads targeting industrial metaverse platforms"
        author = "x7peeps Digital Forensics"
        date = "2026/07/19"
        reference = "Industrial Metaverse Social Engineering"
        mitre_attack = "T1566"
        severity = "high"
    strings:
        $vr1 = "oculus" ascii nocase
        $vr2 = "quest" ascii nocase
        $vr3 = "vr-headset" ascii nocase
        $vr4 = "metaverse" ascii nocase
        $vr5 = "avatar" ascii nocase
        $vr6 = "immersive" ascii nocase
        $phish1 = "login" ascii nocase
        $phish2 = "password" ascii nocase
        $phish3 = "credential" ascii nocase
        $phish4 = "update" ascii nocase
        $mal1 = "powershell" ascii nocase
        $mal2 = "cmd.exe" ascii nocase
        $mal3 = "wscript" ascii nocase
        $mal4 = "mshta" ascii nocase
    condition:
        (2 of ($vr*) and 2 of ($phish*) and 1 of ($mal*))
}