扩展现实XR与元宇宙平台安全取证深度分析

#!/bin/bash
XR_DEVICE_SERIAL=$1
OUTPUT_DIR="./xr_firmware_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "[*] 连接XR设备: $XR_DEVICE_SERIAL"
adb -s "$XR_DEVICE_SERIAL" shell getprop ro.build.display.id > "$OUTPUT_DIR/build_info.txt"
adb -s "$XR_DEVICE_SERIAL" shell getprop ro.build.version.incremental >> "$OUTPUT_DIR/build_info.txt"
adb -s "$XR_DEVICE_SERIAL" shell getprop ro.product.model >> "$OUTPUT_DIR/build_info.txt"
adb -s "$XR_DEVICE_SERIAL" shell getprop ro.product.device >> "$OUTPUT_DIR/build_info.txt"

echo "[*] 提取分区信息..."
adb -s "$XR_DEVICE_SERIAL" shell "cat /proc/partitions" > "$OUTPUT_DIR/partitions.txt"
adb -s "$XR_DEVICE_SERIAL" shell "ls -la /dev/block/by-name/" > "$OUTPUT_DIR/block_devices.txt"

echo "[*] 提取引导加载程序信息..."
adb -s "$XR_DEVICE_SERIAL" shell "cat /proc/cmdline" > "$OUTPUT_DIR/kernel_cmdline.txt"
adb -s "$XR_DEVICE_SERIAL" shell "dmesg | grep -i 'boot\|secure\|verified\|unlock'" > "$OUTPUT_DIR/boot_log.txt"

echo "[*] 检测设备解锁状态..."
UNLOCK_STATE=$(adb -s "$XR_DEVICE_SERIAL" shell "getprop ro.boot.verifiedbootstate" 2>/dev/null)
echo "Verified Boot State: $UNLOCK_STATE" >> "$OUTPUT_DIR/boot_state.txt"
UNLOCK=$(adb -s "$XR_DEVICE_SERIAL" shell "getprop ro.boot.flash.locked" 2>/dev/null)
echo "Flash Lock State: $UNLOCK" >> "$OUTPUT_DIR/boot_state.txt"

echo "[*] 提取系统属性完整列表..."
adb -s "$XR_DEVICE_SERIAL" shell "getprop" > "$OUTPUT_DIR/full_properties.txt"

echo "[*] 提取OTA更新缓存..."
adb -s "$XR_DEVICE_SERIAL" shell "ls -la /data/system/updates/" >> "$OUTPUT_DIR/ota_cache.txt" 2>/dev/null
adb -s "$XR_DEVICE_SERIAL" pull /data/system/updates/ "$OUTPUT_DIR/ota_packages/" 2>/dev/null

echo "[*] 提取已安装应用列表..."
adb -s "$XR_DEVICE_SERIAL" shell "pm list packages -f" > "$OUTPUT_DIR/installed_packages.txt"
adb -s "$XR_DEVICE_SERIAL" shell "dumpsys package" > "$OUTPUT_DIR/package_dumpsys.txt"

echo "[*] 提取系统镜像分区（需要root）..."
for partition in boot system vendor dtbo vbmeta; do
    echo "    提取 ${partition} 分区..."
    adb -s "$XR_DEVICE_SERIAL" shell "dd if=/dev/block/by-name/${partition} of=/sdcard/${partition}.img" 2>/dev/null
    adb -s "$XR_DEVICE_SERIAL" pull "/sdcard/${partition}.img" "$OUTPUT_DIR/${partition}.img" 2>/dev/null
    adb -s "$XR_DEVICE_SERIAL" shell "rm /sdcard/${partition}.img" 2>/dev/null
done

echo "[*] 验证固件完整性..."
for img in "$OUTPUT_DIR"/*.img; do
    if [ -f "$img" ]; then
        MD5=$(md5 -q "$img" 2>/dev/null || md5sum "$img" | awk '{print $1}')
        SHA256=$(shasum -a 256 "$img" 2>/dev/null | awk '{print $1}' || sha256sum "$img" | awk '{print $1}')
        echo "$(basename $img): MD5=$MD5 SHA256=$SHA256" >> "$OUTPUT_DIR/integrity_check.txt"
    fi
done

echo "[+] 固件提取完成，输出目录: $OUTPUT_DIR"
echo "[+] 完整性报告: $OUTPUT_DIR/integrity_check.txt"

import json
import statistics
from datetime import datetime, timedelta
from collections import defaultdict, Counter
from typing import List, Dict, Tuple, Optional

class EyeTrackingForensicAnalyzer:
    FIXATION_THRESHOLD_MS = 100
    SACCADE_VELOCITY_THRESHOLD = 300
    ANOMALY_ZSCORE_THRESHOLD = 2.5
    PRIVACY_SENSITIVE_REGIONS = {
        "keyboard_area": {"desc": "虚拟键盘输入区域", "risk": "密码与输入窃取"},
        "system_ui": {"desc": "系统界面控件", "risk": "权限与设置信息"},
        "avatar_face": {"desc": "其他用户Avatar面部", "risk": "社交关系推断"},
        "content_panel": {"desc": "内容/文档面板", "risk": "阅读内容推断"},
        "passthrough_center": {"desc": "现实环境中心区域", "risk": "环境布局暴露"},
    }

    def __init__(self, gaze_data: List[Dict]):
        self.gaze_data = gaze_data
        self.fixations = []
        self.saccades = []
        self.anomalies = []

    def parse_gaze_stream(self) -> List[Dict]:
        parsed = []
        for entry in self.gaze_data:
            record = {
                "timestamp": datetime.fromisoformat(entry.get("ts", entry.get("timestamp", ""))),
                "x": float(entry.get("x", entry.get("gaze_x", 0))),
                "y": float(entry.get("y", entry.get("gaze_y", 0))),
                "z": float(entry.get("z", entry.get("gaze_z", 0))),
                "confidence": float(entry.get("conf", entry.get("confidence", 0))),
                "pupil_diameter": float(entry.get("pd", entry.get("pupil_diameter", 0))),
                "blink_prob": float(entry.get("blink", entry.get("blink_probability", 0))),
                "session_id": entry.get("session_id", "unknown"),
            }
            parsed.append(record)
        parsed.sort(key=lambda r: r["timestamp"])
        return parsed

    def detect_fixations(self, data: List[Dict]) -> List[Dict]:
        fixations = []
        current_fixation = None

        for i, point in enumerate(data):
            if point["confidence"] < 0.5:
                continue

            if current_fixation is None:
                current_fixation = {
                    "start_idx": i,
                    "end_idx": i,
                    "positions": [(point["x"], point["y"])],
                    "start_time": point["timestamp"],
                    "end_time": point["timestamp"],
                }
                continue

            dx = point["x"] - current_fixation["positions"][-1][0]
            dy = point["y"] - current_fixation["positions"][-1][1]
            distance = (dx**2 + dy**2) ** 0.5

            if distance < self.SACCADE_VELOCITY_THRESHOLD / 60:
                current_fixation["end_idx"] = i
                current_fixation["end_time"] = point["timestamp"]
                current_fixation["positions"].append((point["x"], point["y"]))
            else:
                duration_ms = (current_fixation["end_time"] - current_fixation["start_time"]).total_seconds() * 1000
                if duration_ms >= self.FIXATION_THRESHOLD_MS:
                    avg_x = statistics.mean([p[0] for p in current_fixation["positions"]])
                    avg_y = statistics.mean([p[1] for p in current_fixation["positions"]])
                    current_fixation["centroid"] = (avg_x, avg_y)
                    current_fixation["duration_ms"] = duration_ms
                    fixations.append(current_fixation)
                current_fixation = {
                    "start_idx": i,
                    "end_idx": i,
                    "positions": [(point["x"], point["y"])],
                    "start_time": point["timestamp"],
                    "end_time": point["timestamp"],
                }

        if current_fixation and len(current_fixation["positions"]) > 0:
            duration_ms = (current_fixation["end_time"] - current_fixation["start_time"]).total_seconds() * 1000
            if duration_ms >= self.FIXATION_THRESHOLD_MS:
                avg_x = statistics.mean([p[0] for p in current_fixation["positions"]])
                avg_y = statistics.mean([p[1] for p in current_fixation["positions"]])
                current_fixation["centroid"] = (avg_x, avg_y)
                current_fixation["duration_ms"] = duration_ms
                fixations.append(current_fixation)

        self.fixations = fixations
        return fixations

    def detect_saccades(self, data: List[Dict], fixations: List[Dict]) -> List[Dict]:
        saccades = []
        for i in range(len(fixations) - 1):
            start = fixations[i]
            end = fixations[i + 1]
            dx = end["centroid"][0] - start["centroid"][0]
            dy = end["centroid"][1] - start["centroid"][1]
            amplitude = (dx**2 + dy**2) ** 0.5
            dt = (end["start_time"] - start["end_time"]).total_seconds()
            if dt > 0:
                velocity = amplitude / dt
            else:
                velocity = float("inf")

            saccades.append({
                "start_pos": start["centroid"],
                "end_pos": end["centroid"],
                "amplitude": amplitude,
                "velocity": velocity,
                "start_time": start["end_time"],
                "end_time": end["start_time"],
                "direction": self._calc_direction(dx, dy),
            })
        self.saccades = saccades
        return saccades

    def _calc_direction(self, dx: float, dy: float) -> str:
        import math
        angle = math.degrees(math.atan2(-dy, dx)) % 360
        directions = ["E", "NE", "N", "NW", "W", "SW", "S", "SE"]
        idx = round(angle / 45) % 8
        return directions[idx]

    def analyze_privacy_exposure(self) -> Dict:
        exposure_report = {
            "total_fixation_count": len(self.fixations),
            "total_fixation_duration_ms": sum(f["duration_ms"] for f in self.fixations),
            "region_heatmap": defaultdict(float),
            "longest_fixations": [],
            "sensitive_region_access": [],
            "pupil_diameter_stats": {},
        }

        for fix in self.fixations:
            region = self._classify_region(fix["centroid"])
            exposure_report["region_heatmap"][region] += fix["duration_ms"]

        sorted_regions = sorted(
            exposure_report["region_heatmap"].items(),
            key=lambda x: x[1], reverse=True
        )
        exposure_report["top注视区域"] = sorted_regions[:5]

        sorted_fixations = sorted(self.fixations, key=lambda f: f["duration_ms"], reverse=True)
        exposure_report["longest_fixations"] = [
            {"position": f["centroid"], "duration_ms": f["duration_ms"],
             "region": self._classify_region(f["centroid"])}
            for f in sorted_fixations[:10]
        ]

        for region, info in self.PRIVACY_SENSITIVE_REGIONS.items():
            access_time = exposure_report["region_heatmap"].get(region, 0)
            if access_time > 0:
                exposure_report["sensitive_region_access"].append({
                    "region": region,
                    "description": info["desc"],
                    "risk": info["risk"],
                    "total_duration_ms": access_time,
                })

        return exposure_report

    def _classify_region(self, position: Tuple[float, float]) -> str:
        x, y = position
        if y > 0.7 and 0.3 < x < 0.7:
            return "keyboard_area"
        elif x < 0.2 or x > 0.8:
            return "system_ui"
        elif 0.4 < x < 0.6 and 0.3 < y < 0.6:
            return "avatar_face"
        elif 0.2 < x < 0.8 and 0.2 < y < 0.7:
            return "content_panel"
        else:
            return "passthrough_center"

    def detect_anomalies(self, data: List[Dict]) -> List[Dict]:
        anomalies = []

        durations = [f["duration_ms"] for f in self.fixations]
        if len(durations) > 10:
            mean_dur = statistics.mean(durations)
            stdev_dur = statistics.stdev(durations)
            for fix in self.fixations:
                if stdev_dur > 0:
                    zscore = (fix["duration_ms"] - mean_dur) / stdev_dur
                    if abs(zscore) > self.ANOMALY_ZSCORE_THRESHOLD:
                        anomalies.append({
                            "type": "异常凝视时长",
                            "position": fix["centroid"],
                            "duration_ms": fix["duration_ms"],
                            "zscore": round(zscore, 2),
                            "region": self._classify_region(fix["centroid"]),
                            "timestamp": fix["start_time"].isoformat(),
                        })

        if len(data) > 60:
            window_size = 60
            confidence_values = [d["confidence"] for d in data]
            for i in range(0, len(confidence_values) - window_size, window_size // 2):
                window = confidence_values[i:i + window_size]
                window_mean = statistics.mean(window)
                if window_mean < 0.3:
                    anomalies.append({
                        "type": "低置信度注视区域",
                        "window_start": i,
                        "window_end": i + window_size,
                        "mean_confidence": round(window_mean, 3),
                        "possible_cause": "遮挡/传感器异常/伪造数据",
                    })

        pupil_diameters = [d["pupil_diameter"] for d in data if d["pupil_diameter"] > 0]
        if len(pupil_diameters) > 10:
            pd_mean = statistics.mean(pupil_diameters)
            pd_stdev = statistics.stdev(pupil_diameters)
            for d in data:
                if d["pupil_diameter"] > 0 and pd_stdev > 0:
                    pd_z = (d["pupil_diameter"] - pd_mean) / pd_stdev
                    if abs(pd_z) > 3.0:
                        anomalies.append({
                            "type": "异常瞳孔直径",
                            "diameter_mm": d["pupil_diameter"],
                            "zscore": round(pd_z, 2),
                            "timestamp": d["timestamp"].isoformat(),
                        })

        self.anomalies = anomalies
        return anomalies

    def generate_forensic_report(self) -> Dict:
        data = self.parse_gaze_stream()
        fixations = self.detect_fixations(data)
        saccades = self.detect_saccades(data, fixations)
        privacy_analysis = self.analyze_privacy_exposure()
        anomalies = self.detect_anomalies(data)

        return {
            "analysis_timestamp": datetime.now().isoformat(),
            "data_summary": {
                "total_gaze_points": len(data),
                "time_range": {
                    "start": data[0]["timestamp"].isoformat() if data else None,
                    "end": data[-1]["timestamp"].isoformat() if data else None,
                },
                "sessions_detected": list(set(d["session_id"] for d in data)),
            },
            "fixation_analysis": {
                "total_fixations": len(fixations),
                "mean_fixation_duration_ms": statistics.mean([f["duration_ms"] for f in fixations]) if fixations else 0,
            },
            "saccade_analysis": {
                "total_saccades": len(saccades),
                "mean_amplitude": statistics.mean([s["amplitude"] for s in saccades]) if saccades else 0,
            },
            "privacy_exposure": privacy_analysis,
            "anomalies_detected": anomalies,
            "anomaly_count": len(anomalies),
        }


if __name__ == "__main__":
    sample_data = [
        {"ts": "2026-07-30T10:00:01.000", "x": 0.52, "y": 0.48, "z": -1.0, "conf": 0.95, "pd": 3.8, "blink": 0.0, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:01.008", "x": 0.53, "y": 0.47, "z": -1.0, "conf": 0.93, "pd": 3.82, "blink": 0.0, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:01.016", "x": 0.51, "y": 0.49, "z": -1.0, "conf": 0.96, "pd": 3.85, "blink": 0.0, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:02.000", "x": 0.20, "y": 0.30, "z": -1.0, "conf": 0.88, "pd": 4.10, "blink": 0.0, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:02.500", "x": 0.80, "y": 0.15, "z": -1.0, "conf": 0.72, "pd": 4.30, "blink": 0.0, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:03.000", "x": 0.50, "y": 0.85, "z": -1.0, "conf": 0.91, "pd": 3.90, "blink": 0.0, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:03.100", "x": 0.50, "y": 0.85, "z": -1.0, "conf": 0.92, "pd": 3.88, "blink": 0.0, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:03.200", "x": 0.50, "y": 0.85, "z": -1.0, "conf": 0.90, "pd": 3.87, "blink": 0.0, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:03.500", "x": 0.45, "y": 0.20, "z": -1.0, "conf": 0.15, "pd": 2.50, "blink": 0.8, "session_id": "sess_001"},
        {"ts": "2026-07-30T10:00:04.000", "x": 0.60, "y": 0.50, "z": -1.0, "conf": 0.94, "pd": 6.50, "blink": 0.0, "session_id": "sess_001"},
    ]

    analyzer = EyeTrackingForensicAnalyzer(sample_data)
    report = analyzer.generate_forensic_report()
    print(json.dumps(report, indent=2, ensure_ascii=False, default=str))

import hashlib
import json
from datetime import datetime
from typing import List, Dict, Tuple
import math

class SpatialAnchorForensicAnalyzer:
    MAX_ANCHOR_VELOCITY_MPS = 5.0
    SUSPICIOUS_DENSITY_THRESHOLD = 50
    ANCHOR_DRIFT_TOLERANCE_M = 0.5

    def __init__(self, anchor_logs: List[Dict], environment_map: Dict = None):
        self.anchor_logs = sorted(anchor_logs, key=lambda a: a.get("timestamp", ""))
        self.environment_map = environment_map or {}
        self.alerts = []

    def analyze_anchor_creation_patterns(self) -> Dict:
        creation_times = []
        creator_ids = []
        anchor_positions = []

        for log in self.anchor_logs:
            if log.get("action") == "create":
                creation_times.append(datetime.fromisoformat(log["timestamp"]))
                creator_ids.append(log.get("creator_id", "unknown"))
                anchor_positions.append({
                    "x": log.get("x", 0),
                    "y": log.get("y", 0),
                    "z": log.get("z", 0),
                    "timestamp": log["timestamp"],
                })

        creator_frequency = {}
        for cid in creator_ids:
            creator_frequency[cid] = creator_frequency.get(cid, 0) + 1

        density_map = self._compute_spatial_density(anchor_positions)

        suspicious_creators = []
        for cid, count in creator_frequency.items():
            if count > self.SUSPICIOUS_DENSITY_THRESHOLD:
                suspicious_creators.append({
                    "creator_id": cid,
                    "anchor_count": count,
                    "risk": "高密度锚点注册，可能为锚点投毒攻击",
                })
                self.alerts.append({
                    "type": "ANCHOR_FLOODING",
                    "creator_id": cid,
                    "count": count,
                    "severity": "HIGH",
                })

        return {
            "total_anchors_created": len(anchor_positions),
            "unique_creators": len(creator_frequency),
            "creator_frequency": creator_frequency,
            "suspicious_creators": suspicious_creators,
            "spatial_density": density_map,
        }

    def _compute_spatial_density(self, positions: List[Dict], grid_size: float = 1.0) -> Dict:
        grid = defaultdict(int)
        for pos in positions:
            gx = round(pos["x"] / grid_size) * grid_size
            gy = round(pos["y"] / grid_size) * grid_size
            gz = round(pos["z"] / grid_size) * grid_size
            key = f"{gx:.1f},{gy:.1f},{gz:.1f}"
            grid[key] += 1
        return dict(grid)

    def detect_anchor_drift(self) -> List[Dict]:
        drift_events = []
        for i in range(len(self.anchor_logs) - 1):
            curr = self.anchor_logs[i]
            next_log = self.anchor_logs[i + 1]

            if curr.get("anchor_id") != next_log.get("anchor_id"):
                continue
            if curr.get("action") != "update" or next_log.get("action") != "update":
                continue

            dt = (datetime.fromisoformat(next_log["timestamp"]) - datetime.fromisoformat(curr["timestamp"])).total_seconds()
            if dt <= 0:
                continue

            dx = next_log.get("x", 0) - curr.get("x", 0)
            dy = next_log.get("y", 0) - curr.get("y", 0)
            dz = next_log.get("z", 0) - curr.get("z", 0)
            distance = math.sqrt(dx**2 + dy**2 + dz**2)
            velocity = distance / dt

            if distance > self.ANCHOR_DRIFT_TOLERANCE_M:
                drift_events.append({
                    "anchor_id": curr.get("anchor_id"),
                    "drift_distance_m": round(distance, 4),
                    "velocity_mps": round(velocity, 4),
                    "from_position": {"x": curr.get("x"), "y": curr.get("y"), "z": curr.get("z")},
                    "to_position": {"x": next_log.get("x"), "y": next_log.get("y"), "z": next_log.get("z")},
                    "time_delta_s": round(dt, 3),
                    "timestamp": next_log["timestamp"],
                    "severity": "CRITICAL" if velocity > self.MAX_ANCHOR_VELOCITY_MPS else "MEDIUM",
                })

                if velocity > self.MAX_ANCHOR_VELOCITY_MPS:
                    self.alerts.append({
                        "type": "IMPOSSIBLE_ANCHOR_MOVEMENT",
                        "anchor_id": curr.get("anchor_id"),
                        "velocity_mps": round(velocity, 4),
                        "severity": "CRITICAL",
                    })

        return drift_events

    def detect_unauthorized_anchor_access(self) -> List[Dict]:
        access_violations = []
        for log in self.anchor_logs:
            if log.get("action") in ("delete", "modify"):
                if log.get("creator_id") != log.get("actor_id"):
                    access_violations.append({
                        "anchor_id": log.get("anchor_id"),
                        "action": log["action"],
                        "creator_id": log.get("creator_id"),
                        "actor_id": log.get("actor_id"),
                        "timestamp": log["timestamp"],
                        "violation_type": "跨用户锚点操作",
                    })
                    self.alerts.append({
                        "type": "UNAUTHORIZED_ANCHOR_ACCESS",
                        "anchor_id": log.get("anchor_id"),
                        "actor": log.get("actor_id"),
                        "severity": "HIGH",
                    })
        return access_violations

    def generate_report(self) -> Dict:
        creation_analysis = self.analyze_anchor_creation_patterns()
        drift_events = self.detect_anchor_drift()
        access_violations = self.detect_unauthorized_anchor_access()

        return {
            "analysis_timestamp": datetime.now().isoformat(),
            "total_anchor_events": len(self.anchor_logs),
            "creation_analysis": creation_analysis,
            "drift_events": drift_events,
            "drift_event_count": len(drift_events),
            "access_violations": access_violations,
            "access_violation_count": len(access_violations),
            "alerts": self.alerts,
            "alert_count": len(self.alerts),
        }


if __name__ == "__main__":
    sample_anchors = [
        {"timestamp": "2026-07-30T10:00:00", "action": "create", "anchor_id": "anc_001", "creator_id": "user_a", "actor_id": "user_a", "x": 1.0, "y": 1.5, "z": 2.0},
        {"timestamp": "2026-07-30T10:00:05", "action": "update", "anchor_id": "anc_001", "creator_id": "user_a", "actor_id": "user_a", "x": 1.01, "y": 1.5, "z": 2.0},
        {"timestamp": "2026-07-30T10:00:10", "action": "update", "anchor_id": "anc_001", "creator_id": "user_a", "actor_id": "user_b", "x": 50.0, "y": 1.5, "z": 2.0},
        {"timestamp": "2026-07-30T10:01:00", "action": "delete", "anchor_id": "anc_002", "creator_id": "user_a", "actor_id": "user_c", "x": 3.0, "y": 1.0, "z": 4.0},
    ]
    analyzer = SpatialAnchorForensicAnalyzer(sample_anchors)
    report = analyzer.generate_report()
    print(json.dumps(report, indent=2, ensure_ascii=False, default=str))

import json
from datetime import datetime, timedelta
from collections import defaultdict, Counter
from typing import List, Dict, Optional

class VoiceCommandForensicAnalyzer:
    RAPID_FIRE_THRESHOLD_S = 2.0
    OFF_HOURS_START = 23
    OFF_HOURS_END = 6
    SENSITIVE_COMMANDS = {
        "delete": "删除操作",
        "send": "发送操作",
        "share": "分享操作",
        "purchase": "购买操作",
        "install": "安装操作",
        "settings": "系统设置修改",
        "password": "密码相关操作",
        "unlock": "解锁操作",
        "record": "录制操作",
        "screenshot": "截图操作",
    }
    SUSPICIOUS_PHRASES = [
        "bypass", "override", "admin", "root",
        "disable security", "turn off", "factory reset",
    ]

    def __init__(self, voice_logs: List[Dict]):
        self.voice_logs = sorted(voice_logs, key=lambda v: v.get("timestamp", ""))

    def analyze_temporal_patterns(self) -> Dict:
        temporal_analysis = {
            "hourly_distribution": defaultdict(int),
            "off_hours_commands": [],
            "rapid_fire_events": [],
            "session_analysis": [],
        }

        timestamps = [datetime.fromisoformat(v["timestamp"]) for v in self.voice_logs]

        for ts in timestamps:
            temporal_analysis["hourly_distribution"][ts.hour] += 1

        for log in self.voice_logs:
            ts = datetime.fromisoformat(log["timestamp"])
            if ts.hour >= self.OFF_HOURS_START or ts.hour < self.OFF_HOURS_END:
                temporal_analysis["off_hours_commands"].append({
                    "timestamp": log["timestamp"],
                    "command": log.get("transcript", ""),
                    "confidence": log.get("recognition_confidence", 0),
                })

        for i in range(len(self.voice_logs) - 1):
            ts_curr = datetime.fromisoformat(self.voice_logs[i]["timestamp"])
            ts_next = datetime.fromisoformat(self.voice_logs[i + 1]["timestamp"])
            delta = (ts_next - ts_curr).total_seconds()
            if delta < self.RAPID_FIRE_THRESHOLD_S and delta >= 0:
                temporal_analysis["rapid_fire_events"].append({
                    "event_1": {
                        "timestamp": self.voice_logs[i]["timestamp"],
                        "command": self.voice_logs[i].get("transcript", ""),
                    },
                    "event_2": {
                        "timestamp": self.voice_logs[i + 1]["timestamp"],
                        "command": self.voice_logs[i + 1].get("transcript", ""),
                    },
                    "interval_seconds": round(delta, 3),
                })

        return temporal_analysis

    def analyze_command_content(self) -> Dict:
        content_analysis = {
            "sensitive_commands": [],
            "suspicious_phrases": [],
            "recognition_anomalies": [],
            "command_category_distribution": defaultdict(int),
        }

        for log in self.voice_logs:
            transcript = log.get("transcript", "").lower()

            for keyword, desc in self.SENSITIVE_COMMANDS.items():
                if keyword in transcript:
                    content_analysis["sensitive_commands"].append({
                        "timestamp": log["timestamp"],
                        "transcript": log.get("transcript", ""),
                        "matched_keyword": keyword,
                        "description": desc,
                        "session_id": log.get("session_id", "unknown"),
                    })

            for phrase in self.SUSPICIOUS_PHRASES:
                if phrase in transcript:
                    content_analysis["suspicious_phrases"].append({
                        "timestamp": log["timestamp"],
                        "transcript": log.get("transcript", ""),
                        "matched_phrase": phrase,
                    })

            confidence = log.get("recognition_confidence", 1.0)
            if confidence < 0.3 and len(transcript) > 0:
                content_analysis["recognition_anomalies"].append({
                    "timestamp": log["timestamp"],
                    "transcript": log.get("transcript", ""),
                    "confidence": confidence,
                    "possible_cause": "低置信度语音输入，可能为环境噪音伪造或超声波注入",
                })

            category = log.get("intent_category", "unknown")
            content_analysis["command_category_distribution"][category] += 1

        return content_analysis

    def detect_voice_spoofing_indicators(self) -> List[Dict]:
        spoofing_indicators = []

        confidence_values = [v.get("recognition_confidence", 0) for v in self.voice_logs]
        if len(confidence_values) > 10:
            avg_conf = sum(confidence_values) / len(confidence_values)
            low_conf_runs = 0
            for cv in confidence_values:
                if cv < 0.5:
                    low_conf_runs += 1
                else:
                    if low_conf_runs >= 3:
                        spoofing_indicators.append({
                            "type": "连续低置信度语音输入",
                            "run_length": low_conf_runs,
                            "possible_cause": "语音深度伪造或合成语音",
                        })
                    low_conf_runs = 0

        user_speaking_rate = defaultdict(list)
        for v in self.voice_logs:
            uid = v.get("user_id", "default")
            transcript = v.get("transcript", "")
            if len(transcript) > 0:
                word_count = len(transcript.split())
                duration = v.get("audio_duration_ms", 1000) / 1000
                if duration > 0:
                    user_speaking_rate[uid].append(word_count / duration)

        for uid, rates in user_speaking_rate.items():
            if len(rates) > 5:
                import statistics
                mean_rate = statistics.mean(rates)
                stdev_rate = statistics.stdev(rates)
                for i, rate in enumerate(rates):
                    if stdev_rate > 0 and abs(rate - mean_rate) / stdev_rate > 3.0:
                        spoofing_indicators.append({
                            "type": "异常语速",
                            "user_id": uid,
                            "rate": round(rate, 2),
                            "mean_rate": round(mean_rate, 2),
                            "possible_cause": "合成语音或播放录音",
                        })

        return spoofing_indicators

    def generate_report(self) -> Dict:
        temporal = self.analyze_temporal_patterns()
        content = self.analyze_command_content()
        spoofing = self.detect_voice_spoofing_indicators()

        return {
            "analysis_timestamp": datetime.now().isoformat(),
            "total_voice_events": len(self.voice_logs),
            "temporal_analysis": temporal,
            "content_analysis": content,
            "spoofing_indicators": spoofing,
            "spoofing_indicator_count": len(spoofing),
            "risk_summary": {
                "off_hours_commands": len(temporal.get("off_hours_commands", [])),
                "rapid_fire_events": len(temporal.get("rapid_fire_events", [])),
                "sensitive_commands": len(content.get("sensitive_commands", [])),
                "suspicious_phrases": len(content.get("suspicious_phrases", [])),
            },
        }


if __name__ == "__main__":
    sample_voice = [
        {"timestamp": "2026-07-30T02:15:00", "transcript": "Hey Meta, send message to John", "recognition_confidence": 0.92, "session_id": "vs_001", "user_id": "user_a", "intent_category": "messaging", "audio_duration_ms": 2500},
        {"timestamp": "2026-07-30T02:15:01", "transcript": "install application from web", "recognition_confidence": 0.88, "session_id": "vs_001", "user_id": "user_a", "intent_category": "system", "audio_duration_ms": 3000},
        {"timestamp": "2026-07-30T14:30:00", "transcript": "bypass security check", "recognition_confidence": 0.45, "session_id": "vs_002", "user_id": "unknown", "intent_category": "system", "audio_duration_ms": 2000},
        {"timestamp": "2026-07-30T14:30:01.5", "transcript": "override admin settings", "recognition_confidence": 0.42, "session_id": "vs_002", "user_id": "unknown", "intent_category": "system", "audio_duration_ms": 2200},
    ]

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

#!/bin/bash
XR_NETWORK_INTERFACE=$1
CAPTURE_DIR="./xr_traffic_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$CAPTURE_DIR"

echo "[*] 开始捕获XR设备网络流量..."
echo "[*] 网络接口: $XR_NETWORK_INTERFACE"
echo "[*] 输出目录: $CAPTURE_DIR"

tcpdump -i "$XR_NETWORK_INTERFACE" -w "$CAPTURE_DIR/xr_raw_capture.pcap" -G 300 -W 12 &
TCPDUMP_PID=$!
echo "[*] tcpdump PID: $TCPDUMP_PID"

sleep 10
echo "[*] 正在提取DNS查询..."
tcpdump -r "$CAPTURE_DIR/xr_raw_capture.pcap" -nn port 53 2>/dev/null | \
    grep -oP 'A\?\s+\K[^\s]+' | \
    sort | uniq -c | sort -rn > "$CAPTURE_DIR/dns_queries.txt"

echo "[*] 正在提取TLS SNI信息..."
tcpdump -r "$CAPTURE_DIR/xr_raw_capture.pcap" -nn port 443 -A 2>/dev/null | \
    grep -oP 'Server Name Indication.*?Host Name: \K[^\s]+' | \
    sort | uniq -c | sort -rn > "$CAPTURE_DIR/tls_sni_targets.txt"

echo "[*] 正在识别XR平台API端点..."
tcpdump -r "$CAPTURE_DIR/xr_raw_capture.pcap" -nn port 443 -A 2>/dev/null | \
    grep -iE 'graph\.facebook\.com|oculus\.com|apple\.com|picoxr\.com|microsoft\.com' | \
    head -100 > "$CAPTURE_DIR/xr_api_endpoints.txt"

echo "[*] 正在分析数据量分布..."
tcpdump -r "$CAPTURE_DIR/xr_raw_capture.pcap" -nn 2>/dev/null | \
    awk '{print $3}' | sort | uniq -c | sort -rn | head -20 > "$CAPTURE_DIR/traffic_by_host.txt"

echo "[*] 正在检测异常大流量连接..."
tcpdump -r "$CAPTURE_DIR/xr_raw_capture.pcap" -nn -q 2>/dev/null | \
    awk '{print $3, $5}' | sort | uniq -c | sort -rn | head -20 > "$CAPTURE_DIR/large_transfers.txt"

echo "[*] 停止捕获..."
kill $TCPDUMP_PID 2>/dev/null
wait $TCPDUMP_PID 2>/dev/null

echo "[*] 生成流量摘要..."
TOTAL_PACKETS=$(tcpdump -r "$CAPTURE_DIR/xr_raw_capture.pcap" -nn 2>/dev/null | wc -l)
echo "总数据包数: $TOTAL_PACKETS" > "$CAPTURE_DIR/traffic_summary.txt"
TOTAL_BYTES=$(ls -l "$CAPTURE_DIR/xr_raw_capture.pcap" | awk '{print $5}')
echo "捕获文件大小: $TOTAL_BYTES bytes" >> "$CAPTURE_DIR/traffic_summary.txt"
TCP_COUNT=$(tcpdump -r "$CAPTURE_DIR/xr_raw_capture.pcap" -nn tcp 2>/dev/null | wc -l)
UDP_COUNT=$(tcpdump -r "$CAPTURE_DIR/xr_raw_capture.pcap" -nn udp 2>/dev/null | wc -l)
echo "TCP数据包: $TCP_COUNT" >> "$CAPTURE_DIR/traffic_summary.txt"
echo "UDP数据包: $UDP_COUNT" >> "$CAPTURE_DIR/traffic_summary.txt"

echo "[+] 网络流量分析完成，输出目录: $CAPTURE_DIR"
echo "[+] 流量摘要: $CAPTURE_DIR/traffic_summary.txt"

import json
from datetime import datetime
from typing import List, Dict

class XREvidenceCorrelationEngine:
    SEVERITY_WEIGHTS = {"confirmed_malicious": 10, "highly_suspicious": 6, "needs_attention": 3}
    TIME_CORRELATION_WINDOW_S = 300

    def __init__(self, evidence_items: List[Dict]):
        self.evidence_items = sorted(evidence_items, key=lambda e: e.get("timestamp", ""))

    def build_temporal_clusters(self) -> List[List[Dict]]:
        clusters = []
        current_cluster = []

        for item in self.evidence_items:
            ts = datetime.fromisoformat(item["timestamp"])
            if not current_cluster:
                current_cluster.append(item)
                continue

            last_ts = datetime.fromisoformat(current_cluster[-1]["timestamp"])
            if (ts - last_ts).total_seconds() <= self.TIME_CORRELATION_WINDOW_S:
                current_cluster.append(item)
            else:
                if len(current_cluster) >= 2:
                    clusters.append(current_cluster)
                current_cluster = [item]

        if len(current_cluster) >= 2:
            clusters.append(current_cluster)

        return clusters

    def calculate_attack_chain_probability(self, cluster: List[Dict]) -> Dict:
        total_weight = sum(self.SEVERITY_WEIGHTS.get(e.get("severity", ""), 1) for e in cluster)
        severity_distribution = {}
        for e in cluster:
            sev = e.get("severity", "unknown")
            severity_distribution[sev] = severity_distribution.get(sev, 0) + 1

        technique_ids = list(set(e.get("mitre_technique", "") for e in cluster if e.get("mitre_technique")))
        affected_systems = list(set(e.get("affected_system", "") for e in cluster if e.get("affected_system")))

        if total_weight >= 20:
            confidence = "极高"
        elif total_weight >= 12:
            confidence = "高"
        elif total_weight >= 6:
            confidence = "中"
        else:
            confidence = "低"

        return {
            "cluster_size": len(cluster),
            "time_span": {
                "start": cluster[0]["timestamp"],
                "end": cluster[-1]["timestamp"],
            },
            "total_severity_weight": total_weight,
            "confidence_level": confidence,
            "severity_distribution": severity_distribution,
            "mitre_techniques": technique_ids,
            "affected_systems": affected_systems,
            "evidence_items": [
                {
                    "type": e.get("type", ""),
                    "severity": e.get("severity", ""),
                    "description": e.get("description", ""),
                    "timestamp": e.get("timestamp", ""),
                }
                for e in cluster
            ],
        }

    def correlate(self) -> Dict:
        clusters = self.build_temporal_clusters()
        analyses = [self.calculate_attack_chain_probability(c) for c in clusters]

        confirmed_count = sum(
            1 for a in analyses if a["confidence_level"] in ("极高", "高")
        )

        return {
            "total_evidence_items": len(self.evidence_items),
            "correlated_clusters": len(clusters),
            "cluster_analyses": analyses,
            "confirmed_attack_chains": confirmed_count,
            "recommended_response": (
                "立即启动应急响应" if confirmed_count > 0
                else "持续监控并收集更多证据" if len(clusters) > 0
                else "维持常规监控"
            ),
        }


if __name__ == "__main__":
    evidence = [
        {"timestamp": "2026-07-30T14:00:00", "type": "眼动数据异常导出", "severity": "highly_suspicious", "description": "检测到VR应用大量读取眼动追踪缓存", "mitre_technique": "T1005", "affected_system": "Eye Tracking Service"},
        {"timestamp": "2026-07-30T14:02:30", "type": "网络流量异常", "severity": "confirmed_malicious", "description": "眼动数据通过HTTPS外传至未知名服务器", "mitre_technique": "T1041", "affected_system": "Network Stack"},
        {"timestamp": "2026-07-30T14:05:00", "type": "手势注入", "severity": "confirmed_malicious", "description": "检测到API Hook注入伪造手势", "mitre_technique": "T1059.006", "affected_system": "Hand Tracking API"},
        {"timestamp": "2026-07-30T16:00:00", "type": "固件校验异常", "severity": "needs_attention", "description": "system分区校验和不匹配", "mitre_technique": "", "affected_system": "System Firmware"},
    ]

    engine = XREvidenceCorrelationEngine(evidence)
    result = engine.correlate()
    print(json.dumps(result, indent=2, ensure_ascii=False, default=str))

title: XR设备异常网络活动检测
id: 8f3a7b2c-4d5e-6f78-9a0b-c1d2e3f45678
status: experimental
description: 检测XR头显设备（Meta Quest/Apple Vision Pro/PICO）的异常网络活动，包括向未知服务器发送大量数据、异常DNS查询模式、以及云渲染流量中的数据泄露指标
references:
  - https://developer.oculus.com/resources/overview-networking/
  - https://developer.apple.com/visionos/
author: x7peeps
date: 2026/07/31
tags:
  - attack.exfiltration
  - attack.t1041
  - attack.t1048
  - xr_security
  - vr_forensics
logsource:
  category: proxy
  product: network
detection:
  selection_xr_api_bulk_upload:
    cs-host|contains:
      - 'graph.facebook.com'
      - 'oculus.com'
      - 'oculuscdn.com'
      - 'picoxr.com'
      - 'bytedance.com'
    cs-uri|endswith:
      - '/upload'
      - '/telemetry'
      - '/analytics'
      - '/sync'
    cs-bytes_out|gt: 104857600

  selection_dns_xr_unusual:
    query|contains:
      - 'spatial-data'
      - 'gaze-tracking'
      - 'eye-metrics'
      - 'anchor-sync'
    query|endswith:
      - '.xyz'
      - '.top'
      - '.cc'

  selection_xr_cloud_rendering:
    cs-host|contains:
      - 'cloud-quest'
      - 'remote-render'
      - 'xr-stream'
    src-bytes_out|gt: 52428800

  selection_xr_spatial_data_exfil:
    cs-host|contains:
      - 'pointcloud'
      - 'slam-data'
      - 'room-scan'
    cs-uri|contains:
      - 'upload'
      - 'export'
      - 'backup'

  condition: selection_xr_api_bulk_upload or selection_dns_xr_unusual or selection_xr_cloud_rendering or selection_xr_spatial_data_exfil
  timeframe: 5m
  level: high

falsepositives:
  - 合法的XR平台云渲染服务
  - 系统更新下载
  - 大型VR应用资产下载

fields:
  - cs-host
  - cs-uri
  - cs-bytes_out
  - src-ip
  - user-agent
  - timestamp

#!/bin/bash
set -euo pipefail

DEVICE_SERIAL=$1
CHECKSUM_DB=${2:-"./xr_known_checksums.db"}
REPORT_DIR="./xr_firmware_audit_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$REPORT_DIR"

RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'

log_info() { echo -e "${GREEN}[INFO]${NC} $(date '+%Y-%m-%d %H:%M:%S') $1"; }
log_warn() { echo -e "${YELLOW}[WARN]${NC} $(date '+%Y-%m-%d %H:%M:%S') $1"; }
log_alert() { echo -e "${RED}[ALERT]${NC} $(date '+%Y-%m-%d %H:%M:%S') $1"; }

echo "===========================================" > "$REPORT_DIR/audit_report.txt"
echo "XR设备固件完整性审计报告" >> "$REPORT_DIR/audit_report.txt"
echo "审计时间: $(date '+%Y-%m-%d %H:%M:%S')" >> "$REPORT_DIR/audit_report.txt"
echo "设备序列号: $DEVICE_SERIAL" >> "$REPORT_DIR/audit_report.txt"
echo "===========================================" >> "$REPORT_DIR/audit_report.txt"

log_info "收集设备基本信息..."
BUILD_ID=$(adb -s "$DEVICE_SERIAL" shell getprop ro.build.display.id 2>/dev/null || echo "UNKNOWN")
DEVICE_MODEL=$(adb -s "$DEVICE_SERIAL" shell getprop ro.product.model 2>/dev/null || echo "UNKNOWN")
ANDROID_VER=$(adb -s "$DEVICE_SERIAL" shell getprop ro.build.version.release 2>/dev/null || echo "UNKNOWN")
SECURITY_PATCH=$(adb -s "$DEVICE_SERIAL" shell getprop ro.build.version.security_patch 2>/dev/null || echo "UNKNOWN")

echo "设备型号: $DEVICE_MODEL" >> "$REPORT_DIR/audit_report.txt"
echo "构建ID: $BUILD_ID" >> "$REPORT_DIR/audit_report.txt"
echo "Android版本: $ANDROID_VER" >> "$REPORT_DIR/audit_report.txt"
echo "安全补丁级别: $SECURITY_PATCH" >> "$REPORT_DIR/audit_report.txt"

log_info "检查设备安全启动状态..."
BOOT_STATE=$(adb -s "$DEVICE_SERIAL" shell getprop ro.boot.verifiedbootstate 2>/dev/null || echo "unknown")
FLASH_LOCKED=$(adb -s "$DEVICE_SERIAL" shell getprop ro.boot.flash.locked 2>/dev/null || echo "unknown")
echo "Verified Boot State: $BOOT_STATE" >> "$REPORT_DIR/boot_security.txt"
echo "Flash Lock State: $FLASH_LOCKED" >> "$REPORT_DIR/boot_security.txt"

if [ "$BOOT_STATE" != "green" ]; then
    log_alert "安全启动状态异常: $BOOT_STATE (期望值: green)"
    echo "ALERT: 安全启动状态异常 - $BOOT_STATE" >> "$REPORT_DIR/alerts.txt"
fi

if [ "$FLASH_LOCKED" = "0" ] || [ "$FLASH_LOCKED" = "unlocked" ]; then
    log_alert "设备Bootloader处于解锁状态!"
    echo "ALERT: Bootloader已解锁" >> "$REPORT_DIR/alerts.txt"
fi

log_info "提取关键分区校验信息..."
SYSTEM_HASH=$(adb -s "$DEVICE_SERIAL" shell "sha256sum /dev/block/by-name/system" 2>/dev/null | awk '{print $1}' || echo "EXTRACTION_FAILED")
VENDOR_HASH=$(adb -s "$DEVICE_SERIAL" shell "sha256sum /dev/block/by-name/vendor" 2>/dev/null | awk '{print $1}' || echo "EXTRACTION_FAILED")
BOOT_HASH=$(adb -s "$DEVICE_SERIAL" shell "sha256sum /dev/block/by-name/boot" 2>/dev/null | awk '{print $1}' || echo "EXTRACTION_FAILED")

echo "system分区SHA256: $SYSTEM_HASH" >> "$REPORT_DIR/partition_checksums.txt"
echo "vendor分区SHA256: $VENDOR_HASH" >> "$REPORT_DIR/partition_checksums.txt"
echo "boot分区SHA256: $BOOT_HASH" >> "$REPORT_DIR/partition_checksums.txt"

if [ -f "$CHECKSUM_DB" ]; then
    log_info "对比已知校验和数据库..."
    for PARTITION in system vendor boot; do
        CURRENT_HASH=$(grep "${PARTITION}:" "$REPORT_DIR/partition_checksums.txt" | awk '{print $2}')
        KNOWN_HASH=$(grep "^${DEVICE_MODEL}.*${PARTITION}:" "$CHECKSUM_DB" 2>/dev/null | tail -1 | awk '{print $NF}')
        if [ -n "$KNOWN_HASH" ] && [ "$CURRENT_HASH" != "$KNOWN_HASH" ]; then
            log_alert "${PARTITION}分区校验和不匹配! 当前: ${CURRENT_HASH} 已知: ${KNOWN_HASH}"
            echo "ALERT: ${PARTITION}分区校验和不匹配" >> "$REPORT_DIR/alerts.txt"
        elif [ -n "$KNOWN_HASH" ]; then
            log_info "${PARTITION}分区校验和验证通过"
        fi
    done
fi

log_info "检查已安装应用的完整性..."
adb -s "$DEVICE_SERIAL" shell "pm list packages -f" > "$REPORT_DIR/installed_apps.txt"
adb -s "$DEVICE_SERIAL" shell "dumpsys package" > "$REPORT_DIR/package_details.txt"

SUSPICIOUS_PERMISSIONS="android.permission.READ_EYE_TRACKING android.permission.HAND_TRACKING android.permission.SPATIAL_AUDIO android.permission.CAMERA android.permission.RECORD_AUDIO"
for perm in $SUSPICIOUS_PERMISSIONS; do
    APPS_WITH_PERM=$(adb -s "$DEVICE_SERIAL" shell "dumpsys package" 2>/dev/null | grep -B5 "$perm" | grep "Package \[" | awk '{print $2}' | tr -d ']' || true)
    if [ -n "$APPS_WITH_PERM" ]; then
        echo "权限 $perm 的持有应用:" >> "$REPORT_DIR/permission_audit.txt"
        echo "$APPS_WITH_PERM" >> "$REPORT_DIR/permission_audit.txt"
        echo "---" >> "$REPORT_DIR/permission_audit.txt"
    fi
done

log_info "检查系统日志中的安全事件..."
adb -s "$DEVICE_SERIAL" shell "logcat -d -t 10000 | grep -iE 'root|exploit|hook|xposed|frida|magisk|safety|tamper'" > "$REPORT_DIR/security_events.txt" 2>/dev/null

ROOT_INDICATORS=$(adb -s "$DEVICE_SERIAL" shell "which su 2>/dev/null; ls /system/app/Superuser.apk 2>/dev/null; ls /data/adb/magisk 2>/dev/null" 2>/dev/null || true)
if [ -n "$ROOT_INDICATORS" ]; then
    log_alert "检测到设备已获取root权限!"
    echo "ALERT: 设备已Root" >> "$REPORT_DIR/alerts.txt"
    echo "$ROOT_INDICATORS" >> "$REPORT_DIR/alerts.txt"
fi

ALERT_COUNT=$(wc -l < "$REPORT_DIR/alerts.txt" 2>/dev/null || echo 0)
echo "" >> "$REPORT_DIR/audit_report.txt"
echo "审计结论: 发现 $ALERT_COUNT 个安全告警" >> "$REPORT_DIR/audit_report.txt"

log_info "固件完整性审计完成"
log_info "报告目录: $REPORT_DIR"
echo "[+] 审计报告: $REPORT_DIR/audit_report.txt"
echo "[+] 告警详情: $REPORT_DIR/alerts.txt"

import json
import statistics
from datetime import datetime
from typing import List, Dict, Tuple
import math

class GestureInjectionDetector:
    MAX_PHYSICAL_HAND_VELOCITY = 3.0
    IMPOSSIBLE_ACCELERATION_THRESHOLD = 50.0
    JOINT_ANGLE_PHYSICAL_LIMITS = {
        "thumb": {"min": 0, "max": 90},
        "index": {"min": 0, "max": 120},
        "middle": {"min": 0, "max": 120},
        "ring": {"min": 0, "max": 120},
        "pinky": {"min": 0, "max": 120},
    }
    JITTER_INJECTION_THRESHOLD = 0.05

    def __init__(self, hand_tracking_data: List[Dict]):
        self.hand_data = sorted(hand_tracking_data, key=lambda h: h.get("timestamp", ""))
        self.alerts = []

    def analyze_velocity_anomalies(self) -> List[Dict]:
        anomalies = []
        for i in range(1, len(self.hand_data)):
            prev = self.hand_data[i - 1]
            curr = self.hand_data[i]

            dt = (datetime.fromisoformat(curr["timestamp"]) - datetime.fromisoformat(prev["timestamp"])).total_seconds()
            if dt <= 0:
                continue

            for hand in ["left", "right"]:
                px = prev.get(f"{hand}_wrist_x", 0)
                py = prev.get(f"{hand}_wrist_y", 0)
                pz = prev.get(f"{hand}_wrist_z", 0)
                cx = curr.get(f"{hand}_wrist_x", 0)
                cy = curr.get(f"{hand}_wrist_y", 0)
                cz = curr.get(f"{hand}_wrist_z", 0)

                distance = math.sqrt((cx - px)**2 + (cy - py)**2 + (cz - pz)**2)
                velocity = distance / dt

                if velocity > self.MAX_PHYSICAL_HAND_VELOCITY:
                    anomalies.append({
                        "type": "超物理极限手部速度",
                        "hand": hand,
                        "velocity_ms": round(velocity, 4),
                        "threshold_ms": self.MAX_PHYSICAL_HAND_VELOCITY,
                        "timestamp": curr["timestamp"],
                        "position": {"x": cx, "y": cy, "z": cz},
                        "severity": "CRITICAL",
                        "indication": "手势坐标注入或传感器欺骗",
                    })

            if i >= 2:
                for hand in ["left", "right"]:
                    v1_x = (self.hand_data[i-1].get(f"{hand}_wrist_x", 0) - self.hand_data[i-2].get(f"{hand}_wrist_x", 0)) / max(dt, 0.001)
                    v2_x = (curr.get(f"{hand}_wrist_x", 0) - prev.get(f"{hand}_wrist_x", 0)) / max(dt, 0.001)
                    acceleration = abs(v2_x - v1_x) / max(dt, 0.001)

                    if acceleration > self.IMPOSSIBLE_ACCELERATION_THRESHOLD:
                        anomalies.append({
                            "type": "不可能的手部加速度",
                            "hand": hand,
                            "acceleration": round(acceleration, 4),
                            "timestamp": curr["timestamp"],
                            "severity": "HIGH",
                            "indication": "坐标篡改或注入攻击",
                        })

        return anomalies

    def detect_jitter_injection(self) -> List[Dict]:
        jitter_events = []
        window_size = 10

        for i in range(window_size, len(self.hand_data)):
            window = self.hand_data[i - window_size:i]
            for axis in ["x", "y", "z"]:
                for hand in ["left", "right"]:
                    values = [w.get(f"{hand}_wrist_{axis}", 0) for w in window]
                    if len(values) >= window_size:
                        diffs = [abs(values[j+1] - values[j]) for j in range(len(values)-1)]
                        mean_diff = statistics.mean(diffs) if diffs else 0
                        stdev_diff = statistics.stdev(diffs) if len(diffs) > 1 else 0

                        if mean_diff < self.JITTER_INJECTION_THRESHOLD and stdev_diff > 0:
                            high_freq_count = sum(1 for d in diffs if d > mean_diff + 3 * stdev_diff)
                            if high_freq_count > window_size * 0.3:
                                jitter_events.append({
                                    "type": "高频抖动注入",
                                    "hand": hand,
                                    "axis": axis,
                                    "mean_diff": round(mean_diff, 6),
                                    "stdev_diff": round(stdev_diff, 6),
                                    "high_freq_ratio": round(high_freq_count / len(diffs), 2),
                                    "timestamp": self.hand_data[i]["timestamp"],
                                    "severity": "MEDIUM",
                                })

        return jitter_events

    def detect_phantom_gestures(self) -> List[Dict]:
        phantom_events = []

        for i in range(1, len(self.hand_data)):
            prev = self.hand_data[i - 1]
            curr = self.hand_data[i]

            prev_pinch = prev.get("right_pinch_strength", 0)
            curr_pinch = curr.get("right_pinch_strength", 0)

            if prev_pinch < 0.3 and curr_pinch > 0.8:
                dt = (datetime.fromisoformat(curr["timestamp"]) - datetime.fromisoformat(prev["timestamp"])).total_seconds()
                if dt < 0.05:
                    phantom_events.append({
                        "type": "幽灵捏合手势",
                        "hand": "right",
                        "pinch_change": f"{prev_pinch:.2f} -> {curr_pinch:.2f}",
                        "time_delta_ms": round(dt * 1000, 1),
                        "timestamp": curr["timestamp"],
                        "severity": "HIGH",
                        "indication": "伪造捏合手势（跳过渐进阶段）",
                    })

        return phantom_events

    def generate_report(self) -> Dict:
        velocity_anomalies = self.analyze_velocity_anomalies()
        jitter_events = self.detect_jitter_injection()
        phantom_events = self.detect_phantom_gestures()

        all_anomalies = velocity_anomalies + jitter_events + phantom_events
        critical_count = sum(1 for a in all_anomalies if a.get("severity") == "CRITICAL")
        high_count = sum(1 for a in all_anomalies if a.get("severity") == "HIGH")

        return {
            "analysis_timestamp": datetime.now().isoformat(),
            "total_hand_tracking_frames": len(self.hand_data),
            "velocity_anomalies": velocity_anomalies,
            "jitter_injection_events": jitter_events,
            "phantom_gesture_events": phantom_events,
            "total_anomalies": len(all_anomalies),
            "severity_breakdown": {
                "CRITICAL": critical_count,
                "HIGH": high_count,
                "MEDIUM": len(all_anomalies) - critical_count - high_count,
            },
            "injection_detected": critical_count > 0 or high_count > 2,
        }


if __name__ == "__main__":
    sample_hands = [
        {"timestamp": "2026-07-30T10:00:00.000", "left_wrist_x": 0.3, "left_wrist_y": 0.5, "left_wrist_z": 0.8, "right_wrist_x": 0.7, "right_wrist_y": 0.5, "right_wrist_z": 0.8, "right_pinch_strength": 0.1},
        {"timestamp": "2026-07-30T10:00:00.016", "left_wrist_x": 0.31, "left_wrist_y": 0.51, "left_wrist_z": 0.81, "right_wrist_x": 0.71, "right_wrist_y": 0.50, "right_wrist_z": 0.80, "right_pinch_strength": 0.12},
        {"timestamp": "2026-07-30T10:00:00.033", "left_wrist_x": 0.32, "left_wrist_y": 0.52, "left_wrist_z": 0.82, "right_wrist_x": 0.95, "right_wrist_y": 0.50, "right_wrist_z": 0.80, "right_pinch_strength": 0.15},
        {"timestamp": "2026-07-30T10:00:00.050", "left_wrist_x": 0.33, "left_wrist_y": 0.53, "left_wrist_z": 0.83, "right_wrist_x": 0.72, "right_wrist_y": 0.50, "right_wrist_z": 0.80, "right_pinch_strength": 0.95},
    ]

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

title: XR设备手势注入攻击检测
id: 2b4c8d1e-5f6a-7b8c-9d0e-f1a2b3c4d5e6
status: experimental
description: 检测XR设备上的手势注入攻击指标，包括手部追踪API异常调用、Frida Hook痕迹和异常手势数据流
author: x7peeps
date: 2026/07/31
tags:
  - attack.defense_evasion
  - attack.t1055
  - attack.t1059
  - xr_security
logsource:
  category: process_creation
  product: android
detection:
  selection_frida_xr:
    CommandLine|contains:
      - 'frida'
      - 'frida-server'
      - 'frida-gadget'
    CommandLine|contains:
      - 'hand_tracking'
      - 'eye_tracking'
      - 'gesture'
      - 'skeleton'

  selection_hook_framework_xr:
    CommandLine|contains:
      - 'Xposed'
      - 'Substrate'
      - 'Objection'
    CommandLine|contains:
      - 'com.oculus'
      - 'com.meta'
      - 'com.pico'
      - 'com.apple.vision'

  selection_suspicious_xr_service:
    TargetFileName|contains:
      - 'HandTrackingService'
      - 'GestureRecognition'
      - 'SkeletonRenderer'
    TargetFileName|endswith:
      - '.so'
      - '.dex'

  condition: selection_frida_xr or selection_hook_framework_xr or selection_suspicious_xr_service
  level: critical

fields:
  - CommandLine
  - ParentImage
  - TargetFileName
  - User
  - IntegrityLevel