可穿戴设备与智能穿戴安全取证深度分析

sudo apt install wireshark-qt binwalk ghidra
pip install gattacker btlejuice-core
git clone https://github.com/nccgroup/ufonet.git

import asyncio
from bleak import BleakClient

async def extract_wearable_data(mac_address):
    async with BleakClient(mac_address) as client:
        services = client.services
        for service in services:
            print(f"Service UUID: {service.uuid}")
            for char in service.characteristics:
                try:
                    value = await client.read_gatt_char(char.uuid)
                    print(f"  Characteristic: {char.uuid} = {value.hex()}")
                except Exception as e:
                    print(f"  Characteristic: {char.uuid} = READ_FAILED ({e})")

asyncio.run(extract_wearable_data("AA:BB:CC:DD:EE:FF"))

imazing-cli device-backup --device "Apple Watch" --output ./watch_backup/
sqlite3 watch_backup/Health/healthdb.sqlite ".tables"
sqlite3 watch_backup/Health/healthdb.sqlite "SELECT * FROM SAMPLES WHERE type=37;"

import pandas as pd
import numpy as np

def verify_sensor_integrity(csv_path):
    df = pd.read_csv(csv_path, parse_dates=["timestamp"])
    df["time_diff"] = df["timestamp"].diff().dt.total_seconds()
    
    expected_interval = 1.0
    tolerance = 0.1
    anomalies = df[abs(df["time_diff"] - expected_interval) > tolerance]
    
    sampling_rate = 1.0 / df["time_diff"].median()
    if abs(sampling_rate - 1.0) > 0.05:
        print(f"[WARNING] 异常采样率: {sampling_rate:.2f} Hz (期望: 1.0 Hz)")
    
    heart_rate = df["heart_rate"]
    physio_range = (heart_rate > 20) & (heart_rate < 250)
    outliers = heart_rate[~physio_range]
    if not outliers.empty:
        print(f"[ALERT] {len(outliers)} 个心率值超出生理可能范围")
    
    print(f"[INFO] 时间戳异常点: {len(anomalies)} 个")
    return anomalies

import numpy as np
from scipy.spatial.transform import Rotation

def detect_slam_tampering(imu_data, visual_poses):
    gravity_vectors = []
    for i in range(len(imu_data) - 1):
        accel = imu_data[i]["acceleration"]
        dt = imu_data[i+1]["timestamp"] - imu_data[i]["timestamp"]
        if dt > 0.05:
            continue
        gravity_mag = np.linalg.norm(accel)
        gravity_vectors.append(accel / gravity_mag)
    
    gravity_matrix = np.array(gravity_vectors)
    mean_gravity = np.mean(gravity_matrix, axis=0)
    deviations = np.linalg.norm(gravity_matrix - mean_gravity, axis=1)
    anomalies = np.sum(deviations > 0.3)
    
    if anomalies > len(gravity_matrix) * 0.01:
        print(f"[ALERT] 检测到 {anomalies} 个异常重力矢量，可能为 IMU 数据注入")
    
    for i in range(len(visual_poses) - 1):
        pos1 = visual_poses[i]["position"]
        pos2 = visual_poses[i+1]["position"]
        dt = visual_poses[i+1]["timestamp"] - visual_poses[i]["timestamp"]
        velocity = np.linalg.norm(pos2 - pos1) / max(dt, 0.001)
        if velocity > 10.0:
            print(f"[ALERT] 帧间速度 {velocity:.2f} m/s 超出人类运动极限")
    
    return anomalies

import json
import numpy as np
from datetime import datetime, timedelta

def analyze_eye_tracking_privacy(session_data):
    gaze_points = session_data["gaze_data"]
    timestamps = [p["timestamp"] for p in gaze_points]
    positions = np.array([[p["x"], p["y"]] for p in gaze_points])
    
    fixation_threshold = 0.5
    fixations = []
    current_fixation = [positions[0]]
    
    for i in range(1, len(positions)):
        dist = np.linalg.norm(positions[i] - positions[i-1])
        if dist < fixation_threshold:
            current_fixation.append(positions[i])
        else:
            if len(current_fixation) > 5:
                fixations.append({
                    "center": np.mean(current_fixation, axis=0).tolist(),
                    "duration_ms": len(current_fixation) * 8.3,
                    "count": len(current_fixation)
                })
            current_fixation = [positions[i]]
    
    print(f"检测到 {len(fixations)} 个注视点")
    for fix in fixations:
        if fix["duration_ms"] > 500:
            print(f"  长时间注视: 位置={fix['center']}, 时长={fix['duration_ms']:.1f}ms")

sudo python3 btlejuice -u 192.168.1.100 -i hci0

import struct
from datetime import datetime

def detect_ble_mitm(packet_log):
    alerts = []
    rssi_values = []
    
    for pkt in packet_log:
        rssi = pkt.get("rssi")
        if rssi is not None:
            rssi_values.append((pkt["timestamp"], rssi))
    
    if len(rssi_values) > 10:
        rssi_vals = [v[1] for v in rssi_values]
        mean_rssi = np.mean(rssi_vals)
        std_rssi = np.std(rssi_vals)
        spikes = [(t, v) for t, v in rssi_values if abs(v - mean_rssi) > 3 * std_rssi]
        if spikes:
            alerts.append({
                "type": "RSSI_ANOMALY",
                "severity": "SUSPICIOUS",
                "detail": f"检测到 {len(spikes)} 次 RSSI 异常跳变，可能为 MITM 信号切换",
                "timestamps": [s[0] for s in spikes]
            })
    
    pair_events = [p for p in packet_log if p.get("type") == "SMP_PAIRING_REQ"]
    if len(pair_events) > 3:
        time_span = (pair_events[-1]["timestamp"] - pair_events[0]["timestamp"]).total_seconds()
        if time_span < 3600:
            alerts.append({
                "type": "FREQUENT_REPAIRING",
                "severity": "MALICIOUS",
                "detail": f"{time_span:.0f} 秒内检测到 {len(pair_events)} 次配对请求"
            })
    
    for pkt in packet_log:
        if pkt.get("type") == "ATT_READ_RESP" and pkt.get("service_uuid"):
            if "heart_rate" in pkt["service_uuid"]:
                hr_val = struct.unpack("<B", pkt["value"][:1])[0]
                if hr_val > 250 or hr_val < 15:
                    alerts.append({
                        "type": "IMPOSSIBLE_HEART_RATE",
                        "severity": "CONFIRMED",
                        "detail": f"心率值 {hr_val} bpm 超出生理极限，数据可能被伪造"
                    })
    
    return alerts

openocd -f interface/stlink.cfg -f target/stm32l4x.cfg

openocd> init
openocd> reset halt
openocd> stm32l4x mass_erase
openocd> dump_image firmware_dump.bin 0x08000000 0x100000
openocd> verify_image firmware_dump.bin 0x08000000

binwalk -e firmware.bin
cd _firmware.bin.extracted/
find . -name "*.squashfs" -exec unsquashfs {} \;
strings squashfs-root/usr/bin/app_daemon | grep -i "key\|secret\|password\|token"

import subprocess
import os

def analyze_firmware(firmware_path):
    result = subprocess.run(
        ["binwalk", "-e", firmware_path],
        capture_output=True, text=True
    )
    print(result.stdout)
    
    extract_dir = firmware_path + ".extracted"
    if os.path.exists(extract_dir):
        for root, dirs, files in os.walk(extract_dir):
            for f in files:
                fpath = os.path.join(root, f)
                with open(fpath, "rb") as fp:
                    data = fp.read()
                    keywords = [b"API_KEY", b"SECRET", b"PASSWORD", 
                                b"PRIVATE_KEY", b"HARDWARE_ID"]
                    for kw in keywords:
                        idx = data.find(kw)
                        if idx != -1:
                            context = data[max(0,idx-32):idx+64]
                            print(f"[FOUND] {kw.decode()} in {fpath}: {context}")

import hashlib
import math

def detect_malicious_firmware(firmware_path, known_good_hash=None):
    with open(firmware_path, "rb") as f:
        data = f.read()
    
    sha256_hash = hashlib.sha256(data).hexdigest()
    print(f"固件 SHA256: {sha256_hash}")
    
    if known_good_hash and sha256_hash != known_good_hash:
        print("[CRITICAL] 固件哈希与已知官方版本不匹配，可能为恶意固件")
    
    block_size = 1024
    high_entropy_blocks = []
    for i in range(0, len(data) - block_size, block_size):
        block = data[i:i+block_size]
        entropy = -sum(
            (block.count(byte) / block_size) * math.log2(block.count(byte) / block_size)
            for byte in set(block)
        )
        if entropy > 7.5:
            high_entropy_blocks.append({
                "offset": hex(i),
                "size": block_size,
                "entropy": round(entropy, 3)
            })
    
    if high_entropy_blocks:
        print(f"[WARNING] 发现 {len(high_entropy_blocks)} 个高熵数据块（可能为加密 payload）")
        for block in high_entropy_blocks[:5]:
            print(f"  Offset: {block['offset']}, Entropy: {block['entropy']}")
    
    suspicious_strings = [
        b"nc -e", b"/bin/sh", b"reverse", b"shell",
        b"c2server", b"beacon", b"exfil",
        b"keylogger", b"sniffer", b"rootkit"
    ]
    found = []
    for s in suspicious_strings:
        if s in data:
            found.append(s.decode())
    if found:
        print(f"[CRITICAL] 发现恶意字符串: {found}")
    
    return sha256_hash, high_entropy_blocks, found

import struct

def verify_cgm_data_integrity(cgm_readings):
    alerts = []
    readings = cgm_readings["glucose_values"]
    
    for i, reading in enumerate(readings):
        if reading["value"] < 40 or reading["value"] > 400:
            alerts.append({
                "type": "OUT_OF_RANGE",
                "reading": reading,
                "severity": "CONFIRMED"
            })
        
        if i > 0:
            time_diff = reading["timestamp"] - readings[i-1]["timestamp"]
            value_diff = abs(reading["value"] - readings[i-1]["value"])
            rate = value_diff / max(time_diff / 60, 1)
            if rate > 3.0:
                alerts.append({
                    "type": "ABNORMAL_RATE",
                    "rate_mg_dl_min": rate,
                    "severity": "SUSPICIOUS"
                })
    
    calibrations = cgm_readings.get("calibrations", [])
    for i in range(1, len(calibrations)):
        interval = calibrations[i]["timestamp"] - calibrations[i-1]["timestamp"]
        if interval < 3600:
            alerts.append({
                "type": "FREQUENT_CALIBRATION",
                "interval_seconds": interval,
                "severity": "SUSPICIOUS"
            })
    
    return alerts

import requests
import json
from datetime import datetime

class WearableCloudForensics:
    def __init__(self, access_token):
        self.token = access_token
        self.headers = {"Authorization": f"Bearer {access_token}"}
    
    def extract_daily_activity(self, date_str):
        url = f"https://www.googleapis.com/fitness/v1/users/me/dataSources"
        resp = requests.get(url, headers=self.headers)
        data_sources = resp.json().get("dataSource", [])
        
        activity_data = []
        for ds in data_sources:
            if "com.google.step" in ds.get("dataTypeId", ""):
                dataset_url = (
                    f"https://www.googleapis.com/fitness/v1/users/me/"
                    f"dataSources/{ds['dataSourceId']}/datasets/"
                    f"{date_str}T00:00:00.000Z-{date_str}T23:59:59.999Z"
                )
                ds_resp = requests.get(dataset_url, headers=self.headers)
                activity_data.append(ds_resp.json())
        
        return activity_data
    
    def extract_heart_rate_data(self, date_str):
        url = f"https://www.googleapis.com/fitness/v1/users/me/dataSources"
        resp = requests.get(url, headers=self.headers)
        data_sources = resp.json().get("dataSource", [])
        
        hr_data = []
        for ds in data_sources:
            if "com.google.heart_rate" in ds.get("dataTypeId", ""):
                dataset_url = (
                    f"https://www.googleapis.com/fitness/v1/users/me/"
                    f"dataSources/{ds['dataSourceId']}/datasets/"
                    f"{date_str}T00:00:00.000Z-{date_str}T23:59:59.999Z"
                )
                ds_resp = requests.get(dataset_url, headers=self.headers)
                hr_data.append(ds_resp.json())
        
        return hr_data

    def analyze_cloud_sync_anomalies(self, activity_data):
        anomalies = []
        for source in activity_data:
            for point in source.get("point", []):
                start_ns = int(point["startTimeNanos"])
                end_ns = int(point["EndTimeNanos"])
                duration = (end_ns - start_ns) / 1e9
                
                if duration > 86400:
                    anomalies.append({
                        "type": "EXCESSIVE_DURATION",
                        "start": datetime.fromtimestamp(start_ns / 1e9).isoformat(),
                        "duration_hours": duration / 3600
                    })
                
                sync_gap = int(point["startTimeNanos"]) / 1e9
                if sync_gap > 2592000:
                    anomalies.append({
                        "type": "SYNC_GAP",
                        "gap_days": sync_gap / 86400
                    })
        
        return anomalies

nfc-list
nfc-mfultralight dump nfc_payment_data.dump
strings nfc_payment_data.dump | grep -E "^[0-9]{16}$"
python3 -c "
import struct
with open('nfc_payment_data.dump', 'rb') as f:
    data = f.read()
    for i in range(0, len(data) - 4, 4):
        if data[i] == 0x57 and data[i+1] == 0x07:
            token = data[i+2:i+9].hex()
            print(f'发现支付 Token: {token}')
"

import numpy as np
from scipy import signal

def detect_sensor_forgery(sensor_data, sensor_type="accelerometer"):
    anomalies = []
    
    if sensor_type == "accelerometer":
        x = np.array([d["x"] for d in sensor_data])
        y = np.array([d["y"] for d in sensor_data])
        z = np.array([d["z"] for d in sensor_data])
        
        freqs = np.fft.rfftfreq(len(x), d=1.0/100)
        fft_x = np.abs(np.fft.rfft(x))
        
        peak_freqs = freqs[fft_x > np.mean(fft_x) * 5]
        if len(peak_freqs) > 0:
            dominant_freq = peak_freqs[np.argmax(fft_x[fft_x > np.mean(fft_x) * 5])]
            if 4.8 < dominant_freq < 5.2:
                anomalies.append({
                    "type": "SYNTHETIC_WALKING_PATTERN",
                    "dominant_freq_hz": dominant_freq,
                    "severity": "CONFIRMED",
                    "detail": "检测到精确 5Hz 峰值，可能为合成的步态数据"
                })
        
        magnitude = np.sqrt(x**2 + y**2 + z**2)
        gravity_deviation = abs(np.mean(magnitude) - 9.81)
        if gravity_deviation > 1.5:
            anomalies.append({
                "type": "GRAVITY_DEVIATION",
                "deviation_ms2": gravity_deviation,
                "severity": "HIGHLY_SUSPICIOUS",
                "detail": f"加速度矢量和平均值偏移 {gravity_deviation:.2f} m/s²"
            })
    
    elif sensor_type == "heart_rate":
        hr_values = np.array([d["value"] for d in sensor_data])
        
        if np.std(hr_values) < 1.0 and len(hr_values) > 100:
            anomalies.append({
                "type": "UNNATURAL_CONSTANCY",
                "std_deviation": float(np.std(hr_values)),
                "severity": "CONFIRMED",
                "detail": "心率值标准差过小，不符合生理变异性"
            })
        
        unique_vals = len(np.unique(hr_values))
        if unique_vals < 10 and len(hr_values) > 50:
            anomalies.append({
                "type": "LOW_CARDINALITY",
                "unique_values": unique_vals,
                "severity": "HIGHLY_SUSPICIOUS",
                "detail": f"心率值仅有 {unique_vals} 个不同取值，可能为伪造数据"
            })
    
    return anomalies

title: 可穿戴设备BLE中间人攻击检测
id: d4e5f6a7-8b9c-4d0e-1f2a-3b4c5d6e7f8a
status: experimental
description: 检测可穿戴设备BLE通信中可能的中间人攻击行为
author: x7peeps-blue-team
date: 2026/07/23
tags:
  - attack.technique.t1557
  - attack.tactic.adversary-in-the-middle
  - wearable-security
  - ble-forensics
logsource:
  category: ble_monitoring
  product: wearable_device
detection:
  selection_rssi_anomaly:
    event_type: ble_rssi_change
    rssi_delta_db:
      gte: 20
  selection_repairing:
    event_type: ble_pairing_request
    consecutive_count:
      gte: 3
    time_window_seconds:
      lte: 3600
  selection_connection_param:
    event_type: ble_connection_update
    interval_change_percent:
      gte: 50
  condition: selection_rssi_anomaly or selection_repairing or selection_connection_param
falsepositives:
  - 设备移动导致的正常RSSI变化
  - 多设备环境下的正常重连
level: high

---

title: 可穿戴设备传感器数据伪造检测
id: a1b2c3d4-e5f6-7890-abcd-ef1234567890
status: experimental
description: 检测可穿戴设备传感器数据中的异常模式，可能指示数据伪造
author: x7peeps-blue-team
date: 2026/07/23
tags:
  - attack.technique.t1565
  - attack.tactic.data-manipulation
  - wearable-security
  - sensor-forensics
logsource:
  category: sensor_monitoring
  product: wearable_device
detection:
  selection_hr_out_of_range:
    event_type: heart_rate_reading
    value_bpm:
      lt: 20
    OR:
      value_bpm:
        gt: 250
  selection_accelerometer_anomaly:
    event_type: accelerometer_data
    gravity_deviation_ms2:
      gte: 1.5
  selection_gps_speed_anomaly:
    event_type: gps_track
    speed_kmh:
      gte: 200
  selection_timestamp_anomaly:
    event_type: sensor_reading
    timestamp_regression: true
  condition: selection_hr_out_of_range or selection_accelerometer_anomaly or selection_gps_speed_anomaly or selection_timestamp_anomaly
falsepositives:
  - 传感器硬件故障
  - 极端运动场景下的合法数据
level: high

---

title: 可穿戴设备固件完整性校验失败
id: b3c4d5e6-f7a8-9012-bcde-f12345678901
status: stable
description: 检测可穿戴设备固件更新后的签名或哈希校验失败
author: x7peeps-blue-team
date: 2026/07/23
tags:
  - attack.technique.t1601
  - attack.tactic.patch-system-image
  - wearable-security
  - firmware-integrity
logsource:
  category: firmware_verification
  product: wearable_device
detection:
  selection_sig_fail:
    event_type: firmware_verify
    signature_valid: false
  selection_hash_mismatch:
    event_type: firmware_verify
    hash_match: false
  selection_ota_anomaly:
    event_type: ota_update
    source_ip_external: true
    official_server: false
  condition: selection_sig_fail or selection_hash_mismatch or selection_ota_anomaly
falsepositives:
  - 开发环境下的正常固件更新
level: critical

---

title: 可穿戴支付异常交易检测
id: c4d5e6f7-a8b9-0123-cdef-123456789012
status: experimental
description: 检测可穿戴设备NFC/二维码支付中的异常交易模式
author: x7peeps-blue-team
date: 2026/07/23
tags:
  - attack.technique.t1110
  - attack.tactic.credential-access
  - wearable-security
  - payment-forensics
logsource:
  category: payment_transaction
  product: wearable_device
detection:
  selection_rapid_transactions:
    event_type: nfc_payment
    transaction_count:
      gte: 5
    time_window_minutes:
      lte: 10
  selection_geo_anomaly:
    event_type: nfc_payment
    geo_distance_km:
      gte: 500
    time_diff_minutes:
      lte: 30
  selection_amount_anomaly:
    event_type: nfc_payment
    amount_local_currency:
      gte: 10000
    user_typical_max:
      lte: 1000
  condition: selection_rapid_transactions or selection_geo_anomaly or selection_amount_anomaly
falsepositives:
  - 用户正常高频消费
  - 旅行场景下的合理异地交易
level: high

import asyncio
from bleak import BleakScanner

async def hunt_wearable_threats(scan_duration=30):
    devices = await BleakScanner.discover(timeout=scan_duration)
    
    threat_indicators = []
    for device in devices:
        risk_score = 0
        flags = []
        
        name = device.name or ""
        addr = device.address
        
        known_brands = ["Apple", "Fitbit", "Garmin", "Samsung", 
                        "Huawei", "Xiaomi", "Amazfit", "Withings"]
        is_known = any(b.lower() in name.lower() for b in known_brands)
        
        if device.rssi and device.rssi > -40:
            risk_score += 2
            flags.append("异常强信号（可能为中继攻击）")
        
        addr_type = int(addr.split(":")[0], 16) >> 6
        if addr_type == 0:
            risk_score += 1
            flags.append("静态BLE地址（设备可能禁用RPA）")
        
        if not is_known and any(c in name for c in ["BLE", "Sensor", "Track"]):
            risk_score += 3
            flags.append("可疑设备名称（可能为追踪器/监听设备）")
        
        advertisement = device.metadata.get("service_uuids", [])
        suspicious_services = [
            "0000fe9c-0000-1000-8000-00805f9b34fb",
            "6e400001-b5a3-f393-e0a9-e50e24dcca9e"
        ]
        if any(s in advertisement for s in suspicious_services):
            risk_score += 2
            flags.append("包含可疑BLE服务UUID")
        
        if risk_score > 0:
            threat_indicators.append({
                "device_name": name,
                "address": addr,
                "rssi": device.rssi,
                "risk_score": risk_score,
                "flags": flags,
                "service_uuids": advertisement
            })
    
    threat_indicators.sort(key=lambda x: x["risk_score"], reverse=True)
    for threat in threat_indicators:
        print(f"[RISK SCORE: {threat['risk_score']}] {threat['device_name']} ({threat['address']})")
        for flag in threat["flags"]:
            print(f"  -> {flag}")
    
    return threat_indicators

asyncio.run(hunt_wearable_threats())

#!/bin/bash

Wearable_Forensics_Audit() {
    echo "=== 可穿戴设备安全审计自动化扫描 ==="
    echo "扫描时间: $(date '+%Y-%m-%d %H:%M:%S')"
    
    echo ""
    echo "--- BLE 设备扫描 ---"
    timeout 60 bluetoothctl scan on &
    sleep 30
    bluetoothctl devices | while read line; do
        mac=$(echo "$line" | awk '{print $2}')
        name=$(echo "$line" | cut -d' ' -f3-)
        echo "  设备: $name (MAC: $mac)"
    done
    
    echo ""
    echo "--- 网络连接检查 ---"
    netstat -tlnp 2>/dev/null | grep -E ":(8080|9090|4444|1337|5555)" && \
        echo "  [ALERT] 发现可疑端口监听"
    
    echo ""
    echo "--- USB 设备检查 ---"
    lsusb | grep -iE "ble|bluetooth|wearable|fitbit|garmin|apple" && \
        echo "  [INFO] 发现可穿戴设备 USB 连接"
    
    echo ""
    echo "--- 进程检查 ---"
    ps aux | grep -iE "btlejuice|gattacker|ubertooth|bettercap" | grep -v grep && \
        echo "  [ALERT] 发现 BLE 攻击工具运行中"
    
    echo ""
    echo "--- 日志分析 ---"
    dmesg | grep -iE "usb.*disconnect|ble.*fail|bluetooth.*error" | tail -20
    
    echo ""
    echo "=== 审计完成 ==="
}

Wearable_Forensics_Audit