脑机接口(BCI)安全取证深度分析

pip install mne numpy scipy scikit-learn matplotlib
pip install brainflow
sudo apt install ghidra sigrok pulseview wireshark

import mne
import numpy as np
from scipy import signal
from scipy.stats import kurtosis, skew

def detect_eeg_suspected_forgery(raw_data, sfreq=256, epoch_len=256):
    epochs = np.array([
        raw_data[i:i+epoch_len]
        for i in range(0, len(raw_data) - epoch_len, epoch_len // 2)
    ])
    psd_results = []
    kurtosis_vals = []
    skewness_vals = []
    for epoch in epochs:
        freqs, psd = signal.welch(epoch, fs=sfreq, nperseg=min(256, len(epoch)))
        psd_log = np.log10(psd + 1e-10)
        psd_slope = np.polyfit(freqs[1:], psd_log[1:], 1)[0]
        psd_results.append(psd_slope)
        kurtosis_vals.append(kurtosis(epoch))
        skewness_vals.append(skew(epoch))
    psd_arr = np.array(psd_results)
    kurt_arr = np.array(kurtosis_vals)
    skew_arr = np.array(skewness_vals)
    suspicious_epochs = []
    for idx, (slope, kurt, skew_v) in enumerate(zip(psd_arr, kurt_arr, skew_arr)):
        reasons = []
        if slope > -0.5:
            reasons.append(f"PSD斜率异常平坦({slope:.3f}，正常约-1.0)")
        if abs(kurt) < 0.5:
            reasons.append(f"峰度过低({kurt:.3f}，真实EEG峰度通常>2)")
        if abs(skew_v) < 0.1:
            reasons.append(f"偏度过低({skew_v:.3f}，真实EEG偏度通常>0.3)")
        if reasons:
            suspicious_epochs.append({
                "epoch_index": idx,
                "time_range": f"{idx*epoch_len/sfreq:.2f}-{(idx+1)*epoch_len/sfreq:.2f}s",
                "reasons": reasons
            })
    return suspicious_epochs

test_signal = np.random.randn(25600) * 20
results = detect_eeg_suspected_forgery(test_signal, sfreq=256)
print(f"[!] 检测到 {len(results)} 个可疑时段")
for r in results:
    print(f"  时段 {r['time_range']}: {', '.join(r['reasons'])}")

python3 -c "
import mne
import numpy as np
raw = mne.io.read_raw_edf('suspicious_eeg.edf', preload=True, verbose=False)
data = raw.get_data()
ch_idx = 0
from scipy.signal import stft
f, t, Zxx = stft(data[ch_idx], fs=raw.info['sfreq'], nperseg=128)
power = np.abs(Zxx) ** 2
mean_power_per_freq = np.mean(power, axis=1)
ratio = mean_power_per_freq[5] / (mean_power_per_freq[20] + 1e-10)
print(f'Alpha/Beta功率比: {ratio:.3f}')
if ratio > 5.0:
    print('[!] 警告: Alpha/Beta比值异常偏高，可能为GAN生成伪影')
elif ratio < 0.3:
    print('[!] 警告: Alpha/Beta比值异常偏低，可能为信号注入')
else:
    print('[*] 频谱比值在正常范围内')
"

sudo openocd -f interface/jlink.cfg -f target/stm32f4x.cfg -c "init; halt; dump_image bci_firmware.bin 0x08000000 0x100000; resume; exit"
binwalk -e bci_firmware.bin
strings -n 8 bci_firmware.bin | grep -iE "api|endpoint|password|key|secret|token"

binwalk --dd="squashfs filesystem:squashfs" firmware.bin
binwalk --dd="uImage:uImage" firmware.bin
file _firmware.bin.extracted/*
strings _firmware.bin.extracted/squashfs-root/usr/bin/bci_processor | head -50

objdump -d bci_processor | grep -A 20 "<ble_packet_handler>"
objdump -d bci_processor | grep -A 30 "<eeg_decode_model>"
objdump -d bci_processor | grep -A 15 "<firmware_update_handler>"

sha256sum official_firmware.bin extracted_firmware.bin
diff <(xxd official_firmware.bin) <(xxd extracted_firmware.bin) | head -100
radiff2 -s official_firmware.bin extracted_firmware.bin

import sqlite3
import json
import os

def extract_bci_app_data(db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchall()
    eeg_tables = []
    for table in tables:
        table_name = table[0]
        cursor.execute(f"PRAGMA table_info({table_name})")
        columns = [col[1] for col in cursor.fetchall()]
        eeg_keywords = ['eeg', 'brain', 'signal', 'neural', 'channel', 'electrode', 'sample']
        if any(kw in col.lower() for col in columns for kw in eeg_keywords):
            eeg_tables.append((table_name, columns))
    extracted_data = []
    for table_name, columns in eeg_tables:
        cursor.execute(f"SELECT * FROM {table_name}")
        rows = cursor.fetchall()
        for row in rows:
            row_dict = dict(zip(columns, row))
            extracted_data.append(row_dict)
    conn.close()
    return eeg_tables, extracted_data

def scan_bci_log_files(log_dir):
    suspicious_entries = []
    keywords = ['eeg_data', 'neural_signal', 'brain_pattern', 'upload', 'sync']
    for root, dirs, files in os.walk(log_dir):
        for f in files:
            fpath = os.path.join(root, f)
            try:
                with open(fpath, 'r', errors='ignore') as fh:
                    for line_no, line in enumerate(fh, 1):
                        if any(kw in line.lower() for kw in keywords):
                            suspicious_entries.append({
                                "file": fpath,
                                "line": line_no,
                                "content": line.strip()[:200]
                            })
            except Exception:
                continue
    return suspicious_entries

results = scan_bci_log_files('/var/log')
print(f"[*] 扫描完成，发现 {len(results)} 条可疑日志条目")

sudo btmgmt --index 0 le on
sudo btmgmt --index 0 bredr off
sudo hcitool lescan --passive --duplicates
sudo btlejuice-proxy -i hci0 -w hci1 -l 0.0.0.0

sudo python3 -m btlejuice -u
tshark -i hci0 -w bci_ble_capture.pcap -f "btle"
tshark -r bci_ble_capture.pcap -Y "btle.length > 0" -T fields -e frame.time -e btle.advertising_address -e btle.length

pip-audit --desc -f json > bci_dependency_audit.json
safety check --json > bci_safety_report.json
trivy fs --scanners vuln --format json -o bci_trivy_report.json .

title: BCI设备异常BLE连接频率检测
id: 7f3a2b1c-8d4e-5f6a-9c1b-2e3d4f5a6b7c
status: experimental
description: 检测BCI设备在短时间内出现异常频繁的BLE连接建立事件，可能表明BLE劫持或中间人攻击尝试
references:
  - https://attack.mitre.org/techniques/T1557/
  - https://attack.mitre.org/techniques/T1571/
author: BCI Security Forensics Team
date: 2026/07/30
tags:
  - attack.credential_access
  - attack.t1557
  - attack.t1571
  - bci.security
logsource:
  product: bci
  service: bci_bluetooth
detection:
  selection_bci_connection:
    - EventID: 4688
      CommandLine|contains|all:
        - 'gatttool'
        - '-b'
    - EventID: 5156
      DestinationPort: 78
      DestinationPort: 79
  timeframe_30min:
    EventID: 5156
  condition: selection_bci_connection and timeframe_30min | count() by TargetProcessName > 20
level: high
falsepositives:
  - BCI设备固件更新时的正常多连接
  - 多用户环境下的正常设备扫描

title: BCI设备固件完整性校验失败
id: 8a4b3c2d-9e0f-1a2b-3c4d-5e6f7a8b9c0d
status: experimental
description: 检测BCI设备在启动或OTA更新过程中固件完整性校验失败事件，可能表明固件被篡改或供应链攻击
references:
  - https://attack.mitre.org/techniques/T1601/
  - https://attack.mitre.org/techniques/T1195/
author: BCI Security Forensics Team
date: 2026/07/30
tags:
  - attack.impact
  - attack.t1601
  - attack.t1195
  - bci.security
logsource:
  product: bci
  service: bci_firmware_verification
detection:
  selection_firmware_fail:
    - EventID: 1001
      Level: 2
      Message|contains:
        - 'firmware'
        - 'integrity'
        - 'failed'
    - EventID: 1001
      Message|contains:
        - 'signature'
        - 'verification'
        - 'error'
    - EventID: 1001
      Message|contains:
        - 'hash'
        - 'mismatch'
  condition: selection_firmware_fail
level: critical
falsepositives:
  - 固件更新过程中的网络中断导致的临时校验失败
  - 首次刷写非官方开发固件（需在开发环境中排除）

#!/bin/bash
BCI_FIRMWARE_DIR="/opt/bci_firmware"
KNOWN_HASH_DB="/opt/bci_firmware/known_hashes.sha256"
ALERT_LOG="/var/log/bci_firmware_integrity.log"
ALERT_EMAIL="bci-security@example.com"

echo "[$(date '+%Y-%m-%d %H:%M:%S')] BCI固件完整性校验启动" >> "$ALERT_LOG"

if [ ! -f "$KNOWN_HASH_DB" ]; then
    echo "[ERROR] 哈希数据库不存在: $KNOWN_HASH_DB" >> "$ALERT_LOG"
    exit 1
fi

VIOLATIONS=0

while IFS=' ' read -r expected_hash filepath; do
    if [ ! -f "$filepath" ]; then
        echo "[ALERT] 固件文件缺失: $filepath" >> "$ALERT_LOG"
        VIOLATIONS=$((VIOLATIONS + 1))
        continue
    fi
    actual_hash=$(sha256sum "$filepath" | awk '{print $1}')
    if [ "$actual_hash" != "$expected_hash" ]; then
        echo "[CRITICAL] 固件完整性校验失败: $filepath" >> "$ALERT_LOG"
        echo "  期望哈希: $expected_hash" >> "$ALERT_LOG"
        echo "  实际哈希: $actual_hash" >> "$ALERT_LOG"
        echo "  文件大小: $(stat -f%z "$filepath" 2>/dev/null || stat -c%s "$filepath" 2>/dev/null) bytes" >> "$ALERT_LOG"
        echo "  修改时间: $(stat -f%m "$filepath" 2>/dev/null || stat -c%Y "$filepath" 2>/dev/null)" >> "$ALERT_LOG"
        VIOLATIONS=$((VIOLATIONS + 1))
    else
        echo "[OK] $filepath 完整性校验通过" >> "$ALERT_LOG"
    fi
done < "$KNOWN_HASH_DB"

if [ $VIOLATIONS -gt 0 ]; then
    echo "[ALERT] 发现 $VIOLATIONS 个固件完整性违规" >> "$ALERT_LOG"
    if command -v mail &> /dev/null; then
        tail -20 "$ALERT_LOG" | mail -s "[BCI-CRITICAL] 固件完整性校验失败 - $(hostname)" "$ALERT_EMAIL"
    fi
else
    echo "[INFO] 所有固件完整性校验通过" >> "$ALERT_LOG"
fi

find "$BCI_FIRMWARE_DIR" -name "*.bin" -newer "$KNOWN_HASH_DB" -exec echo "[WARNING] 新增未记录固件: {}" \; >> "$ALERT_LOG"
echo "[$(date '+%Y-%m-%d %H:%M:%S')] BCI固件完整性校验完成" >> "$ALERT_LOG"

import numpy as np
from scipy import signal
from scipy.stats import kurtosis, entropy
import json
import sys

SFREQ = 256
BANDS = {
    "delta": (0.5, 4),
    "theta": (4, 8),
    "alpha": (8, 13),
    "beta": (13, 30),
    "gamma": (30, 100)
}

def compute_band_powers(data, sfreq=SFREQ):
    freqs, psd = signal.welch(data, fs=sfreq, nperseg=min(512, len(data)))
    band_powers = {}
    total_power = np.trapz(psd, freqs)
    for band_name, (low, high) in BANDS.items():
        mask = (freqs >= low) & (freqs <= high)
        band_powers[band_name] = np.trapz(psd[mask], freqs[mask]) / total_power
    return band_powers

def detect_temporal_anomalies(data, sfreq=SFREQ, window_sec=2, threshold_std=3.0):
    window_samples = int(window_sec * sfreq)
    n_windows = len(data) // window_samples
    if n_windows < 3:
        return []
    window_stds = []
    for i in range(n_windows):
        chunk = data[i * window_samples:(i + 1) * window_samples]
        window_stds.append(np.std(chunk))
    mean_std = np.mean(window_stds)
    std_std = np.std(window_stds)
    anomalies = []
    for i, ws in enumerate(window_stds):
        z_score = (ws - mean_std) / (std_std + 1e-10)
        if abs(z_score) > threshold_std:
            anomalies.append({
                "window": i,
                "time_range": f"{i * window_sec:.1f}-{(i + 1) * window_sec:.1f}s",
                "std_value": float(ws),
                "z_score": float(z_score),
                "type": "energy_surge" if z_score > 0 else "energy_drop"
            })
    return anomalies

def detect_spectral_anomalies(data, sfreq=SFREQ):
    freqs, psd = signal.welch(data, fs=sfreq, nperseg=min(512, len(data)))
    psd_log = np.log10(psd + 1e-10)
    slope = np.polyfit(freqs[1:], psd_log[1:], 1)[0]
    alpha_mask = (freqs >= 8) & (freqs <= 13)
    beta_mask = (freqs >= 13) & (freqs <= 30)
    alpha_power = np.mean(psd[alpha_mask]) if np.any(alpha_mask) else 0
    beta_power = np.mean(psd[beta_mask]) if np.any(beta_mask) else 0
    ab_ratio = alpha_power / (beta_power + 1e-10)
    peak_freq = freqs[np.argmax(psd[1:]) + 1]
    anomalies = []
    if slope > -0.3:
        anomalies.append({
            "type": "flat_spectrum",
            "description": f"PSD斜率异常平坦({slope:.3f}，正常约-1.0)",
            "severity": "high"
        })
    if ab_ratio > 8.0:
        anomalies.append({
            "type": "excessive_alpha",
            "description": f"Alpha/Beta比值异常偏高({ab_ratio:.2f})",
            "severity": "medium"
        })
    sharp_peaks = np.sum(psd[1:] > 5 * np.median(psd[1:]))
    if sharp_peaks > 3:
        anomalies.append({
            "type": "narrowband_interference",
            "description": f"检测到{sharp_peaks}个异常窄带尖峰",
            "severity": "medium"
        })
    return anomalies, slope, ab_ratio, peak_freq

def detect_injection_patterns(data, sfreq=SFREQ):
    channel_data = data if data.ndim == 1 else data[0]
    anomalies = detect_temporal_anomalies(channel_data, sfreq)
    spectral, slope, ab_ratio, peak_freq = detect_spectral_anomalies(channel_data)
    band_powers = compute_band_powers(channel_data, sfreq)
    kurtosis_val = kurtosis(channel_data)
    psd_freqs, psd_vals = signal.welch(channel_data, fs=sfreq, nperseg=min(512, len(channel_data)))
    psd_normalized = psd_vals / (np.sum(psd_vals) + 1e-10)
    spectral_entropy = entropy(psd_normalized)
    report = {
        "analysis_timestamp": str(np.datetime64('now')),
        "sampling_rate": sfreq,
        "duration_seconds": len(channel_data) / sfreq,
        "signal_statistics": {
            "mean": float(np.mean(channel_data)),
            "std": float(np.std(channel_data)),
            "kurtosis": float(kurtosis_val),
            "peak_to_peak": float(np.ptp(channel_data))
        },
        "band_powers": {k: float(v) for k, v in band_powers.items()},
        "spectral_features": {
            "psd_slope": float(slope),
            "alpha_beta_ratio": float(ab_ratio),
            "peak_frequency_hz": float(peak_freq),
            "spectral_entropy": float(spectral_entropy)
        },
        "temporal_anomalies": anomalies,
        "spectral_anomalies": spectral,
        "risk_assessment": {
            "overall_risk": "low",
            "confidence": 0.0,
            "indicators": []
        }
    }
    risk_score = 0
    indicators = []
    if len(anomalies) > 0:
        risk_score += len(anomalies) * 15
        indicators.append(f"时间域异常: {len(anomalies)}个")
    if slope > -0.3:
        risk_score += 30
        indicators.append("频谱斜率异常")
    if kurtosis_val < 1.0:
        risk_score += 25
        indicators.append(f"峰度过低({kurtosis_val:.2f})")
    if spectral_entropy < 2.0:
        risk_score += 20
        indicators.append("频谱熵异常偏低")
    report["risk_assessment"]["indicators"] = indicators
    if risk_score >= 60:
        report["risk_assessment"]["overall_risk"] = "critical"
    elif risk_score >= 40:
        report["risk_assessment"]["overall_risk"] = "high"
    elif risk_score >= 20:
        report["risk_assessment"]["overall_risk"] = "medium"
    else:
        report["risk_assessment"]["overall_risk"] = "low"
    report["risk_assessment"]["risk_score"] = risk_score
    report["risk_assessment"]["confidence"] = min(risk_score / 100.0, 0.95)
    return report

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <eeg_data.npy> [channel_index]")
        print("生成示例数据进行演示...")
        demo_data = np.random.randn(25600) * 20 + 5 * np.sin(2 * np.pi * 10 * np.arange(25600) / 256)
        demo_data += np.random.uniform(-2, 2, 25600)
        result = detect_injection_patterns(demo_data)
        print(json.dumps(result, indent=2, ensure_ascii=False))
        sys.exit(0)
    data_path = sys.argv[1]
    ch_idx = int(sys.argv[2]) if len(sys.argv) > 2 else 0
    data = np.load(data_path)
    if data.ndim > 1:
        data = data[ch_idx]
    result = detect_injection_patterns(data)
    print(json.dumps(result, indent=2, ensure_ascii=False))

# BLE设备MAC地址模式（需在具体设备上验证）
# nRF Sniffer格式
btle.advertising_address == "XX:XX:XX:XX:XX:XX" && btle.advertising_header.pdu_type == 0x00
# EEG数据流异常模式
# 在BLE GATT写操作中检测非预期数据包序列
btle.opcode == 0x12 && btle.handle == 0x000X && btle.value[0:4] != expected_eeg_header

# BrainCo FocusCalm BLE GATT服务UUID
service_uuid: "0000ffe0-0000-1000-8000-00805f9b34fb"
# XOR加密密钥（固定值）
encryption_key: 0xA5
# API IDOR漏洞端点模式
api_pattern: "https://api.brainco.com/v1/users/{user_id}/eeg-data"
# 异常BLE流量特征
ble_characteristic_handle: 0x000E
ble_write_without_auth: true