自主无人机与集群安全取证深度分析

#!/usr/bin/env python3
import numpy as np
import struct
from datetime import datetime

GPS_L1_FREQ = 1575.42e6
SAMPLE_RATE = 2.6e6
C_A_CODE_CHIP = 1023

def generate_ca_code(prn):
    g1 = np.ones(10, dtype=int)
    g2 = np.ones(10, dtype=int)
    g2_taps = {
        1: [2, 6], 2: [3, 7], 3: [4, 8], 4: [5, 9], 5: [1, 9],
        6: [2, 10], 7: [1, 8], 8: [2, 9], 9: [3, 10], 10: [2, 3],
        11: [3, 4], 12: [5, 6], 13: [6, 7], 14: [7, 8], 15: [8, 9],
        16: [9, 10], 17: [1, 4], 18: [2, 5], 19: [3, 6], 20: [4, 7],
        21: [5, 8], 22: [6, 9], 23: [1, 3], 24: [4, 6], 25: [5, 7],
        26: [6, 8], 27: [7, 9], 28: [8, 10], 29: [1, 6], 30: [2, 7],
        31: [3, 8], 32: [4, 9]
    }
    code = np.zeros(C_A_CODE_CHIP, dtype=int)
    for i in range(C_A_CODE_CHIP):
        code[i] = g1[9] ^ g2[g2_taps[prn][0] - 1] ^ g2[g2_taps[prn][1] - 1]
        g1_new = g1[9] ^ g1[2]
        g1 = np.roll(g1, 1)
        g1[0] = g1_new
        g2_new = g2[9] ^ g2[1] ^ g2[2] ^ g2[5]
        g2 = np.roll(g2, 1)
        g2[0] = g2_new
    return 2 * code - 1

def generate_spoofing_signal(target_lat, target_lon, target_alt, duration_s):
    num_samples = int(SAMPLE_RATE * duration_s)
    signal = np.zeros(num_samples, dtype=np.complex128)
    gps_time = datetime.utcnow()
    for prn in range(1, 10):
        ca_code = generate_ca_code(prn)
        repeated_code = np.tile(ca_code, int(SAMPLE_RATE / 1000) + 1)
        chip_rate = int(SAMPLE_RATE / C_A_CODE_CHIP)
        sampled_code = repeated_code[::chip_rate][:num_samples]
        code_bipolar = sampled_code[:num_samples]
        code_bipolar = np.where(code_bipolar > 0, 1.0, -1.0)
        carrier = np.exp(1j * 2 * np.pi * GPS_L1_FREQ / SAMPLE_RATE
                         * np.arange(num_samples))
        amplitude = 1.0 / 9.0
        signal += amplitude * code_bipolar * carrier
    signal = signal / np.max(np.abs(signal)) * 0.8
    return signal.astype(np.complex64)

if __name__ == "__main__":
    spoof_signal = generate_spoofing_signal(
        target_lat=39.9042, target_lon=116.4074, target_alt=100.0,
        duration_s=60
    )
    print(f"Generated spoofing signal: {len(spoof_signal)} samples")
    print(f"Duration: {len(spoof_signal) / SAMPLE_RATE:.1f}s")
    print(f"Target: 39.9042N, 116.4074E, 100m")

#!/bin/bash
SDR_DEVICE="hackrf=0"
CENTER_FREQ=1575420000
SAMPLE_RATE=2000000
GAIN=40
DURATION=30
OUTPUT_DIR="/forensics/gnss/jamming"
THRESHOLD_DB=-80

mkdir -p "$OUTPUT_DIR"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)

rtl_sdr -f "$CENTER_FREQ" -s "$SAMPLE_RATE" -g "$GAIN" \
    -n "$((SAMPLE_RATE * DURATION))" \
    "$OUTPUT_DIR/gnss_capture_${TIMESTAMP}.bin"

fft_data="$OUTPUT_DIR/gnss_capture_${TIMESTAMP}.bin"

python3 - <<PYEOF
import numpy as np
import sys

def detect_jamming(bin_file, sample_rate, threshold_db):
    data = np.fromfile(bin_file, dtype=np.complex64)
    nfft = 4096
    num_segments = len(data) // nfft
    noise_floor = []

    for i in range(num_segments):
        segment = data[i * nfft:(i + 1) * nfft]
        spectrum = np.fft.fftshift(np.fft.fft(segment, nfft))
        psd = 10 * np.log10(np.abs(spectrum)**2 + 1e-10)
        noise_floor.append(np.median(psd))

    avg_noise = np.mean(noise_floor)
    std_noise = np.std(noise_floor)
    anomalies = []
    for i, nf in enumerate(noise_floor):
        if nf > avg_noise + 3 * std_noise:
            anomalies.append((i, nf))
        if nf < avg_noise - 3 * std_noise:
            anomalies.append((i, nf))

    print(f"Average noise floor: {avg_noise:.2f} dB")
    print(f"Std deviation: {std_noise:.2f} dB")
    print(f"Anomalous segments: {len(anomalies)}")
    if anomalies:
        for idx, val in anomalies:
            timestamp = idx * nfft / sample_rate
            print(f"  Segment {idx} at {timestamp:.3f}s: {val:.2f} dB")
    return len(anomalies)

bin_file = sys.argv[1]
sample_rate = int(sys.argv[2])
anomaly_count = detect_jamming(bin_file, sample_rate, ${THRESHOLD_DB})
if anomaly_count > 0:
    print("WARNING: Possible GNSS jamming detected!")
    sys.exit(1)
else:
    print("No jamming patterns detected.")
    sys.exit(0)
PYEOF

#!/usr/bin/env python3
import struct
import json
import sys
from collections import defaultdict
import math

class GNSSFraudDetector:
    def __init__(self, log_path):
        self.log_path = log_path
        self.gps_data = []
        self.imu_data = []
        self.attitude_data = []
        self.alerts = []

    def parse_ulg_header(self, f):
        magic = f.read(8)
        if magic != b"ULog\x01\x00\x00\x00":
            raise ValueError("Not a valid ULog file")
        header_size = struct.unpack("<Q", f.read(8))[0]
        f.read(header_size - 16)

    def parse_ulg_messages(self, f):
        msg_data = b""
        while True:
            sync = f.read(1)
            if not sync or sync != b"\x56":
                break
            msg_size = struct.unpack("<H", f.read(2))[0]
            msg_id = struct.unpack("<B", f.read(1))[0]
            timestamp = struct.unpack("<Q", f.read(8))[0]
            payload = f.read(msg_size - 9)
            if msg_id == 10:
                self._parse_gps_message(timestamp, payload)
            elif msg_id == 7:
                self._parse_imu_message(timestamp, payload)
            elif msg_id == 4:
                self._parse_attitude_message(timestamp, payload)

    def _parse_gps_message(self, ts, payload):
        if len(payload) < 52:
            return
        time_utc = struct.unpack("<Q", payload[0:8])[0]
        lat = struct.unpack("<d", payload[8:16])[0]
        lon = struct.unpack("<d", payload[16:24])[0]
        alt = struct.unpack("<f", payload[24:28])[0]
        eph = struct.unpack("<f", payload[28:32])[0]
        epv = struct.unpack("<f", payload[32:36])[0]
        hdg = struct.unpack("<f", payload[36:40])[0]
        vel_n = struct.unpack("<f", payload[40:44])[0]
        vel_e = struct.unpack("<f", payload[44:48])[0]
        vel_d = struct.unpack("<f", payload[48:52])[0]
        fix_type = struct.unpack("<B", payload[52:53])[0] if len(payload) > 52 else 0
        num_satellites = struct.unpack("<B", payload[53:54])[0] if len(payload) > 53 else 0
        self.gps_data.append({
            "timestamp": ts, "lat": lat, "lon": lon, "alt": alt,
            "eph": eph, "epv": epv, "hdg": hdg,
            "vel_n": vel_n, "vel_e": vel_e, "vel_d": vel_d,
            "fix_type": fix_type, "num_satellites": num_satellites
        })

    def _parse_imu_message(self, ts, payload):
        if len(payload) < 28:
            return
        accel_x = struct.unpack("<f", payload[0:4])[0]
        accel_y = struct.unpack("<f", payload[4:8])[0]
        accel_z = struct.unpack("<f", payload[8:12])[0]
        gyro_x = struct.unpack("<f", payload[12:16])[0]
        gyro_y = struct.unpack("<f", payload[16:20])[0]
        gyro_z = struct.unpack("<f", payload[20:24])[0]
        self.imu_data.append({
            "timestamp": ts,
            "accel": [accel_x, accel_y, accel_z],
            "gyro": [gyro_x, gyro_y, gyro_z]
        })

    def _parse_attitude_message(self, ts, payload):
        if len(payload) < 12:
            return
        roll = struct.unpack("<f", payload[0:4])[0]
        pitch = struct.unpack("<f", payload[4:8])[0]
        yaw = struct.unpack("<f", payload[8:12])[0]
        self.attitude_data.append({
            "timestamp": ts, "roll": roll, "pitch": pitch, "yaw": yaw
        })

    def detect_position_jump(self, threshold_meters=50):
        if len(self.gps_data) < 2:
            return
        for i in range(1, len(self.gps_data)):
            prev = self.gps_data[i - 1]
            curr = self.gps_data[i]
            dlat = (curr["lat"] - prev["lat"]) * 111320
            dlon = (curr["lon"] - prev["lon"]) * 111320 * math.cos(math.radians(curr["lat"]))
            dist = math.sqrt(dlat**2 + dlon**2)
            time_diff = (curr["timestamp"] - prev["timestamp"]) / 1e6
            if time_diff > 0 and dist / time_diff > 100:
                self.alerts.append({
                    "type": "POSITION_JUMP",
                    "severity": "HIGH",
                    "timestamp": curr["timestamp"],
                    "details": f"Position jump: {dist:.1f}m in {time_diff:.3f}s"
                })

    def detect_satellite_anomaly(self):
        if len(self.gps_data) < 10:
            return
        sats = [d["num_satellites"] for d in self.gps_data]
        avg_sat = sum(sats) / len(sats)
        for i, d in enumerate(self.gps_data):
            if d["num_satellites"] > avg_sat * 2 and d["num_satellites"] >= 12:
                self.alerts.append({
                    "type": "SATELLITE_ANOMALY",
                    "severity": "MEDIUM",
                    "timestamp": d["timestamp"],
                    "details": f"Unusual satellite count: {d['num_satellites']} (avg: {avg_sat:.1f})"
                })

    def detect_gnss_imu_divergence(self, threshold=20):
        if len(self.gps_data) < 2 or len(self.imu_data) < 2:
            return
        gps_idx = 0
        for imu in self.imu_data:
            while gps_idx < len(self.gps_data) - 1:
                if self.gps_data[gps_idx + 1]["timestamp"] > imu["timestamp"]:
                    break
                gps_idx += 1
            if gps_idx >= len(self.gps_data) - 1:
                break
            imu_speed = math.sqrt(sum(a**2 for a in imu["accel"][:2]))
            gps_speed = math.sqrt(
                self.gps_data[gps_idx]["vel_n"]**2 +
                self.gps_data[gps_idx]["vel_e"]**2
            )
            if abs(imu_speed - gps_speed) > threshold:
                self.alerts.append({
                    "type": "GNSS_IMU_DIVERGENCE",
                    "severity": "HIGH",
                    "timestamp": imu["timestamp"],
                    "details": f"IMU speed {imu_speed:.2f} vs GPS speed {gps_speed:.2f}"
                })

    def detect_healthy_eph_jump(self, threshold=5.0):
        for i in range(1, len(self.gps_data)):
            prev = self.gps_data[i - 1]
            curr = self.gps_data[i]
            if curr["eph"] < prev["eph"] / 3 and curr["eph"] < 1.0:
                self.alerts.append({
                    "type": "EPH_IMPROVEMENT",
                    "severity": "LOW",
                    "timestamp": curr["timestamp"],
                    "details": f"Eph improved from {prev['eph']:.2f} to {curr['eph']:.2f} (possible spoofing)"
                })

    def generate_report(self):
        report = {
            "file": self.log_path,
            "gps_points": len(self.gps_data),
            "imu_points": len(self.imu_data),
            "total_alerts": len(self.alerts),
            "high_severity": sum(1 for a in self.alerts if a["severity"] == "HIGH"),
            "medium_severity": sum(1 for a in self.gps_data if False),
            "alerts": self.alerts
        }
        return report

    def analyze(self):
        with open(self.log_path, "rb") as f:
            self.parse_ulg_header(f)
            self.parse_ulg_messages(f)
        self.detect_position_jump()
        self.detect_satellite_anomaly()
        self.detect_gnss_imu_divergence()
        self.detect_healthy_eph_jump()
        return self.generate_report()

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <path_to.ulg>")
        sys.exit(1)
    detector = GNSSFraudDetector(sys.argv[1])
    report = detector.analyze()
    print(json.dumps(report, indent=2, ensure_ascii=False))
    if report["high_severity"] > 0:
        print(f"\n[!] {report['high_severity']} HIGH severity alerts - GPS spoofing likely!")
        sys.exit(2)
    elif report["total_alerts"] > 0:
        print(f"\n[*] {report['total_alerts']} alerts detected - manual review recommended")
        sys.exit(1)
    else:
        print("\n[+] No GPS spoofing indicators detected")
        sys.exit(0)

#!/usr/bin/env python3
import struct
import time
import json
import sys
from collections import defaultdict

MAVLINK_V2_HEADER = 0xFD
MAVLINK_V1_HEADER = 0xFE
CMD_NAMES = {
    22: "NAV_WAYPOINT", 31: "NAV_LOITER_UNLIM", 73: "NAV_RETURN_TO_LAUNCH",
    176: "MAV_CMD_COMPONENT_ARM_DISARM", 178: "MAV_CMD_COMPONENT_DISARM",
    204: "NAV_LAND", 21: "NAV_TAKEOFF", 112: "MAV_CMD_START_CALIBRATION",
    241: "MAV_CMD_INJECT_FAILURE", 400: "MAV_CMD_USER_1",
    310: "OBLIQUE_PAYLOAD_DEPLOY", 520: "MAV_CMD_SET_MESSAGE_INTERVAL",
    530: "MAV_CMD_REQUEST_PROTOCOL_VERSION"
}

class MAVLinkForensicAnalyzer:
    def __init__(self):
        self.messages = []
        self.systems = defaultdict(int)
        self.command_history = []
        self.anomalies = []

    def parse_v2_header(self, data, offset):
        if data[offset] != MAVLINK_V2_HEADER:
            return None, 0
        length = data[offset + 1]
        incompat_flags = data[offset + 2]
        compat_flags = data[offset + 3]
        seq = data[offset + 4]
        sysid = data[offset + 5]
        compid = data[offset + 6]
        msgid = (data[offset + 9] << 16) | (data[offset + 8] << 8) | data[offset + 7]
        header_size = 10
        return {
            "version": 2, "length": length, "seq": seq,
            "sysid": sysid, "compid": compid, "msgid": msgid,
            "incompat_flags": incompat_flags
        }, header_size

    def parse_v1_header(self, data, offset):
        if data[offset] != MAVLINK_V1_HEADER:
            return None, 0
        length = data[offset + 1]
        seq = data[offset + 2]
        sysid = data[offset + 3]
        compid = data[offset + 4]
        msgid = data[offset + 5]
        header_size = 6
        return {
            "version": 1, "length": length, "seq": seq,
            "sysid": sysid, "compid": compid, "msgid": msgid
        }, header_size

    def extract_command_payload(self, header, data, offset):
        payload_start = offset + header["version"] + header["length"] + 2
        if header["msgid"] in [176, 178]:
            if payload_start + 6 <= len(data):
                param1 = struct.unpack("<f", data[payload_start:payload_start + 4])[0]
                param2 = struct.unpack("<f", data[payload_start + 4:payload_start + 8])[0]
                return {"param1": param1, "param2": param2}
        return {}

    def detect_unauthorized_commands(self):
        sensitive_cmds = [176, 178, 112, 241, 310]
        for msg in self.messages:
            if msg["msgid"] in sensitive_cmds:
                self.anomalies.append({
                    "type": "SENSITIVE_COMMAND",
                    "sysid": msg["sysid"],
                    "msgid": msg["msgid"],
                    "cmd_name": CMD_NAMES.get(msg["msgid"], "UNKNOWN")
                })

    def detect_system冒充(self):
        sys_profiles = defaultdict(lambda: {"count": 0, "first_seen": 0, "last_seen": 0})
        for msg in self.messages:
            key = f"{msg['sysid']}:{msg['compid']}"
            sys_profiles[key]["count"] += 1
            if sys_profiles[key]["first_seen"] == 0:
                sys_profiles[key]["first_seen"] = msg["timestamp"]
            sys_profiles[key]["last_seen"] = msg["timestamp"]
        for sys_id, profile in sys_profiles.items():
            if profile["count"] == 1:
                self.anomalies.append({
                    "type": "SINGLE_MESSAGE_SYSTEM",
                    "sysid": sys_id,
                    "details": f"System sent only 1 message"
                })

    def generate_report(self):
        return {
            "total_messages": len(self.messages),
            "unique_systems": len(set(f"{m['sysid']}:{m['compid']}" for m in self.messages)),
            "sensitive_commands": len([a for a in self.anomalies if a["type"] == "SENSITIVE_COMMAND"]),
            "anomalies": self.anomalies
        }

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <mavlink_capture.bin>")
        sys.exit(1)
    analyzer = MAVLinkForensicAnalyzer()
    analyzer.detect_unauthorized_commands()
    report = analyzer.generate_report()
    print(json.dumps(report, indent=2, ensure_ascii=False))

#!/bin/bash
FIRMWARE_IMAGE=$1
OUTPUT_DIR="/forensics/firmware"
EXTRACT_DIR="$OUTPUT_DIR/extracted"

mkdir -p "$EXTRACT_DIR"

echo "[+] Firmware hash calculation"
sha256sum "$FIRMWARE_IMAGE" | tee "$OUTPUT_DIR/firmware.sha256"
md5sum "$FIRMWARE_IMAGE" | tee "$OUTPUT_DIR/firmware.md5"

echo "[+] Firmware entropy analysis"
binwalk -E "$FIRMWARE_IMAGE"

echo "[+] Firmware signature extraction"
binwalk -A "$FIRMWARE_IMAGE"

echo "[+] Firmware extraction"
binwalk -e -C "$EXTRACT_DIR" "$FIRMWARE_IMAGE"

echo "[+] Filesystem analysis"
find "$EXTRACT_DIR" -name "passwd" -o -name "shadow" -o -name "*.key" -o -name "*.pem" 2>/dev/null

echo "[+] Hardcoded credential search"
grep -r "password" "$EXTRACT_DIR" --include="*.conf" --include="*.cfg" -i 2>/dev/null
grep -r "api_key" "$EXTRACT_DIR" --include="*.json" --include="*.xml" -i 2>/dev/null
grep -r "secret" "$EXTRACT_DIR" --include="*.conf" --include="*.h" -i 2>/dev/null

echo "[+] Network configuration analysis"
find "$EXTRACT_DIR" -name "*.conf" -exec grep -l "socket\|connect\|bind\|listen" {} \; 2>/dev/null

echo "[+] Backdoor indicator search"
grep -r "nc -l\|ncat\|reverse\|shell\|exec(" "$EXTRACT_DIR" -i 2>/dev/null
grep -r "curl.*\|wget.*" "$EXTRACT_DIR" --include="*.sh" --include="*.py" 2>/dev/null

echo "[+] Binary analysis for ARM/x86"
find "$EXTRACT_DIR" -type f -executable | while read bin; do
    file_type=$(file "$bin" 2>/dev/null)
    if echo "$file_type" | grep -q "ELF"; then
        echo "ELF binary found: $bin"
        echo "  $file_type"
        checksec --file="$bin" 2>/dev/null || true
    fi
done

echo "[+] Strings analysis for suspicious patterns"
strings "$FIRMWARE_IMAGE" | grep -iE "(curl|wget|nc |ncat|/bin/sh|/bin/bash|eval\(|exec\(|system\()" | head -50

echo "[+] Firmware timestamp analysis"
strings "$FIRMWARE_IMAGE" | grep -E "[0-9]{4}-[0-9]{2}-[0-9]{2}" | sort -u | head -20

echo "[+] Analysis complete. Results saved to $OUTPUT_DIR"

#!/usr/bin/env python3
import hashlib
import json
import os
import sys
import subprocess
from pathlib import Path

class FirmwareIntegrityVerifier:
    EXPECTED_BOOTLOADER_MAGIC = bytes([0x7F, 0x45, 0x4C, 0x46])
    PX4_SIGNATURE_OFFSET = 0x1000
    ARDUPILOT_SIGNATURE_MARKER = b"ArduPilot"

    def __init__(self, firmware_path, reference_hashes=None):
        self.firmware_path = Path(firmware_path)
        self.reference_hashes = reference_hashes or {}
        self.results = {
            "file": str(self.firmware_path),
            "checks": [],
            "integrity": True
        }

    def calculate_hashes(self):
        hashes = {}
        with open(self.firmware_path, "rb") as f:
            data = f.read()
        hashes["sha256"] = hashlib.sha256(data).hexdigest()
        hashes["sha1"] = hashlib.sha1(data).hexdigest()
        hashes["md5"] = hashlib.md5(data).hexdigest()
        hashes["size"] = len(data)
        return hashes, data

    def verify_elf_header(self, data):
        if data[:4] == self.EXPECTED_BOOTLOADER_MAGIC:
            return True, "Valid ELF header detected"
        if data[:2] == b"MZ":
            return True, "PE binary detected (Windows GCS)"
        return False, "Unknown binary format"

    def check_reference_hash(self, computed_hashes):
        if not self.reference_hashes:
            return True, "No reference hashes provided"
        for algo, expected in self.reference_hashes.items():
            if algo in computed_hashes:
                if computed_hashes[algo] != expected:
                    return False, f"Hash mismatch: {algo} computed={computed_hashes[algo]} expected={expected}"
        return True, "All hashes match reference"

    def scan_for_backdoor_indicators(self, data):
        indicators = []
        dangerous_patterns = [
            (b"/bin/sh", "Shell execution"),
            (b"/bin/bash", "Bash execution"),
            (b"nc -l", "Netcat listener"),
            (b"ncat", "Ncat connection"),
            (b"reverse_shell", "Reverse shell"),
            (b"curl http", "External HTTP request"),
            (b"wget http", "External HTTP download"),
            (b"base64 -d", "Base64 decode (obfuscation)"),
            (b"exec(", "Python exec call"),
            (b"system(", "C system call"),
            (b"telnetd", "Telnet daemon"),
        ]
        for pattern, desc in dangerous_patterns:
            if pattern in data:
                indicators.append({"pattern": pattern.decode(), "description": desc})
        return indicators

    def verify_compile_timestamp(self, data):
        import struct
        if data[:4] != self.EXPECTED_BOOTLOADER_MAGIC:
            return None, "Non-ELF binary"
        try:
            ts_offset = 16
            timestamp = struct.unpack("<I", data[ts_offset:ts_offset + 4])[0]
            import datetime
            dt = datetime.datetime.fromtimestamp(timestamp)
            return dt.isoformat(), "Compile timestamp extracted"
        except Exception:
            return None, "Cannot extract timestamp"

    def run(self):
        hashes, data = self.calculate_hashes()
        elf_valid, elf_msg = self.verify_elf_header(data)
        self.results["checks"].append({"name": "ELF_HEADER", "passed": elf_valid, "details": elf_msg})
        hash_valid, hash_msg = self.check_reference_hash(hashes)
        self.results["checks"].append({"name": "HASH_VERIFICATION", "passed": hash_valid, "details": hash_msg})
        if not hash_valid:
            self.results["integrity"] = False
        backdoors = self.scan_for_backdoor_indicators(data)
        bd_passed = len(backdoors) == 0
        self.results["checks"].append({"name": "BACKDOOR_SCAN", "passed": bd_passed, "details": f"{len(backdoors)} indicators found", "indicators": backdoors})
        if not bd_passed:
            self.results["integrity"] = False
        ts, ts_msg = self.verify_compile_timestamp(data)
        self.results["checks"].append({"name": "COMPILE_TIMESTAMP", "passed": ts is not None, "details": ts_msg})
        self.results["computed_hashes"] = hashes
        return self.results

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <firmware.bin> [reference_hashes.json]")
        sys.exit(1)
    ref_hashes = {}
    if len(sys.argv) > 2:
        with open(sys.argv[2]) as f:
            ref_hashes = json.load(f)
    verifier = FirmwareIntegrityVerifier(sys.argv[1], ref_hashes)
    result = verifier.run()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    sys.exit(0 if result["integrity"] else 1)

#!/usr/bin/env python3
import hashlib
import json
import time
import random
from collections import defaultdict
from typing import Dict, List, Tuple

class SwarmConsensusSimulator:
    def __init__(self, num_nodes: int, num_byzantine: int):
        self.num_nodes = num_nodes
        self.num_byzantine = num_byzantine
        self.normal_nodes = list(range(num_byzantine, num_nodes))
        self.byzantine_nodes = list(range(num_byzantine))
        self.consensus_rounds = []
        self.anomalies_detected = []

    def simulate_normal_proposal(self, node_id: int, true_value: float) -> Dict:
        noise = random.gauss(0, 0.01)
        return {
            "node_id": node_id,
            "proposed_value": true_value + noise,
            "is_byzantine": False,
            "round": len(self.consensus_rounds),
            "timestamp": time.time(),
            "signature": hashlib.sha256(f"{node_id}:{true_value}".encode()).hexdigest()[:16]
        }

    def simulate_byzantine_proposal(self, node_id: int, true_value: float,
                                     attack_strategy: str = "random") -> Dict:
        if attack_strategy == "random":
            malicious_value = true_value + random.uniform(-100, 100)
        elif attack_strategy == "flip":
            malicious_value = -true_value
        elif attack_strategy == "gradual":
            malicious_value = true_value + random.uniform(-0.5, 0.5)
        else:
            malicious_value = random.uniform(-1000, 1000)
        return {
            "node_id": node_id,
            "proposed_value": malicious_value,
            "is_byzantine": True,
            "attack_strategy": attack_strategy,
            "round": len(self.consensus_rounds),
            "timestamp": time.time(),
            "signature": hashlib.sha256(f"{node_id}:{malicious_value}".encode()).hexdigest()[:16]
        }

    def run_consensus_round(self, true_value: float, attack_strategy: str = "random") -> Dict:
        proposals = []
        for node in self.normal_nodes:
            proposals.append(self.simulate_normal_proposal(node, true_value))
        for node in self.byzantine_nodes:
            proposals.append(self.simulate_byzantine_proposal(node, true_value, attack_strategy))
        valid_proposals = [p for p in proposals if not p["is_byzantine"]]
        all_proposals = proposals
        avg_valid = sum(p["proposed_value"] for p in valid_proposals) / len(valid_proposals)
        avg_all = sum(p["proposed_value"] for p in all_proposals) / len(all_proposals)
        consensus_result = avg_all
        deviation = abs(consensus_result - true_value)
        consensus_achieved = deviation < 0.1
        round_data = {
            "round_id": len(self.consensus_rounds),
            "true_value": true_value,
            "consensus_result": consensus_result,
            "deviation": deviation,
            "consensus_achieved": consensus_achieved,
            "num_proposals": len(proposals),
            "num_byzantine": self.num_byzantine,
            "attack_strategy": attack_strategy,
            "avg_valid": avg_valid,
            "avg_all": avg_all,
            "proposals": proposals
        }
        self.consensus_rounds.append(round_data)
        if not consensus_achieved:
            self.anomalies_detected.append({
                "round": len(self.consensus_rounds) - 1,
                "deviation": deviation,
                "attack": attack_strategy
            })
        return round_data

    def detect_byzantine_nodes(self) -> List[Dict]:
        detection_results = []
        node_history = defaultdict(list)
        for round_data in self.consensus_rounds:
            for proposal in round_data["proposals"]:
                node_history[proposal["node_id"]].append(proposal["proposed_value"])
        for node_id, values in node_history.items():
            if len(values) < 2:
                continue
            mean_val = sum(values) / len(values)
            variance = sum((v - mean_val) ** 2 for v in values) / len(values)
            if variance > 1.0:
                detection_results.append({
                    "node_id": node_id,
                    "mean_proposal": mean_val,
                    "variance": variance,
                    "num_rounds": len(values),
                    "likely_byzantine": True
                })
        return detection_results

    def generate_forensic_report(self) -> Dict:
        byzantine_nodes_detected = self.detect_byzantine_nodes()
        total_rounds = len(self.consensus_rounds)
        failed_rounds = len(self.anomalies_detected)
        return {
            "total_nodes": self.num_nodes,
            "byzantine_injected": self.num_byzantine,
            "byzantine_ratio": self.num_byzantine / self.num_nodes,
            "bft_threshold_exceeded": self.num_byzantine > self.num_nodes / 3,
            "total_rounds": total_rounds,
            "failed_consensus_rounds": failed_rounds,
            "detection_rate": failed_rounds / total_rounds if total_rounds > 0 else 0,
            "detected_byzantine_nodes": byzantine_nodes_detected,
            "anomalies": self.anomalies_detected
        }

if __name__ == "__main__":
    sim = SwarmConsensusSimulator(num_nodes=12, num_byzantine=4)
    print(f"[*] Swarm: {sim.num_nodes} nodes, {sim.num_byzantine} Byzantine")
    print(f"[*] BFT threshold: > {sim.num_nodes // 3} malicious nodes needed")
    for i in range(50):
        true_value = random.uniform(10.0, 20.0)
        strategy = random.choice(["random", "flip", "gradual"])
        result = sim.run_consensus_round(true_value, strategy)
    report = sim.generate_forensic_report()
    print(json.dumps(report, indent=2, ensure_ascii=False))

#!/bin/bash
GCS_LOG_DIR=$1
FORENSICS_OUTPUT="/forensics/gcs"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)

mkdir -p "$FORENSICS_OUTPUT/$TIMESTAMP"

echo "[+] Collecting GCS logs from $GCS_LOG_DIR"
find "$GCS_LOG_DIR" -name "*.log" -o -name "*.ulg" -o -name "*.tlog" \
    -o -name "*.param" -o -name "*.plan" 2>/dev/null | \
    while read -r logfile; do
        dest="$FORENSICS_OUTPUT/$TIMESTAMP/$(basename "$logfile")"
        cp "$logfile" "$dest" 2>/dev/null
        echo "  Collected: $logfile ($(stat -f%z "$logfile" 2>/dev/null || stat -c%s "$logfile" 2>/dev/null) bytes)"
    done

echo "[+] Searching for MAVLink parameter modifications"
find "$FORENSICS_OUTPUT/$TIMESTAMP" -name "*.param" | while read -r param_file; do
    echo "  Parameter file: $param_file"
    grep -i "ARM\|DISARM\|RTL\|FAILSAFE\|AUTO\|MANUAL\|GUIDED" "$param_file" 2>/dev/null
done

echo "[+] Analyzing mission plans for unauthorized waypoints"
find "$FORENSICS_OUTPUT/$TIMESTAMP" -name "*.plan" | while read -r plan_file; do
    echo "  Mission plan: $plan_file"
    python3 -c "
import xml.etree.ElementTree as ET
tree = ET.parse('$plan_file')
root = tree.getroot()
for wp in root.iter('item'):
    lat = wp.find('Lat')
    lon = wp.find('Lon')
    alt = wp.find('Alt')
    if lat is not None and lon is not None:
        print(f'    Waypoint: {lat.text}, {lon.text}, alt={alt.text}')
" 2>/dev/null
done

echo "[+] Checking for suspicious network connections"
netstat -tlnp 2>/dev/null | grep -E "(5760|5762|14550|14551)" | \
    awk '{print "  " $4 " " $7}' >> "$FORENSICS_OUTPUT/$TIMESTAMP/network_connections.txt"

echo "[+] Extracting operator identity from system artifacts"
whoami > "$FORENSICS_OUTPUT/$TIMESTAMP/operator.txt"
echo "Hostname: $(hostname)" >> "$FORENSICS_OUTPUT/$TIMESTAMP/operator.txt"
last -n 10 >> "$FORENSICS_OUTPUT/$TIMESTAMP/operator.txt" 2>/dev/null

echo "[+] GCS forensics collection complete: $FORENSICS_OUTPUT/$TIMESTAMP"

#!/usr/bin/env python3
import json
import sys
from datetime import datetime, timedelta
from collections import defaultdict

class CloudAPILogAnalyzer:
    def __init__(self, log_file):
        self.log_file = log_file
        self.events = []
        self.anomalies = []

    def load_cloud_logs(self):
        with open(self.log_file) as f:
            for line in f:
                try:
                    event = json.loads(line.strip())
                    self.events.append(event)
                except json.JSONDecodeError:
                    continue

    def detect_unauthorized_command(self):
        for event in self.events:
            if event.get("event_type") == "device_command":
                if event.get("source") not in ["authorized_operator", "system"]:
                    self.anomalies.append({
                        "type": "UNAUTHORIZED_COMMAND",
                        "severity": "HIGH",
                        "timestamp": event.get("timestamp"),
                        "device_id": event.get("device_id"),
                        "command": event.get("command"),
                        "source": event.get("source")
                    })

    def detect_api_key_abuse(self):
        api_calls = defaultdict(list)
        for event in self.events:
            if event.get("event_type") == "api_call":
                key = event.get("api_key_id", "unknown")
                api_calls[key].append(event)
        for key_id, calls in api_calls.items():
            timestamps = [c["timestamp"] for c in calls if "timestamp" in c]
            if len(timestamps) > 100:
                self.anomalies.append({
                    "type": "API_KEY_ABUSE",
                    "severity": "MEDIUM",
                    "api_key_id": key_id,
                    "call_count": len(calls),
                    "details": "Excessive API calls detected"
                })

    def detect_telemetry_exfiltration(self):
        telemetry_events = [e for e in self.events if e.get("event_type") == "telemetry_upload"]
        size_threshold = 10 * 1024 * 1024
        for event in telemetry_events:
            if event.get("payload_size", 0) > size_threshold:
                self.anomalies.append({
                    "type": "TELEMETRY_EXFILTRATION",
                    "severity": "MEDIUM",
                    "device_id": event.get("device_id"),
                    "payload_size": event.get("payload_size"),
                    "details": "Abnormally large telemetry upload"
                })

    def generate_report(self):
        return {
            "total_events": len(self.events),
            "total_anomalies": len(self.anomalies),
            "high_severity": sum(1 for a in self.anomalies if a["severity"] == "HIGH"),
            "anomalies": self.anomalies
        }

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <cloud_log.jsonl>")
        sys.exit(1)
    analyzer = CloudAPILogAnalyzer(sys.argv[1])
    analyzer.load_cloud_logs()
    analyzer.detect_unauthorized_command()
    analyzer.detect_api_key_abuse()
    analyzer.detect_telemetry_exfiltration()
    report = analyzer.generate_report()
    print(json.dumps(report, indent=2, ensure_ascii=False))
    sys.exit(1 if report["high_severity"] > 0 else 0)

title: GPS Spoofing Detection via Telemetry Anomaly
id: d4f5e6a7-8b9c-4d0e-1f2a-3b4c5d6e7f8a
status: experimental
description: Detects potential GPS spoofing attacks by analyzing UAS telemetry data for position-velocity inconsistency
author: UAS Security Forensics
date: 2026/07/31
tags:
    - attack.defense_evasion
    - attack.t1562.001
    - uas.gps_spoofing
    - forensics.gnss
logsource:
    category: uas_telemetry
    product: px4
    service: flight_log
detection:
    position_jump:
        selection_gps_jump:
            telemetry_type: GPS_POSITION
            speed_mps: '>100'
    imu_divergence:
        selection_imu_div:
            telemetry_type: IMU_DATA
            accel_xy_sum: '>15'
    satellite_anomaly:
        selection_sat_anomaly:
            telemetry_type: GPS_STATUS
            num_satellites: '>12'
            hdop: '<0.5'
    eph_jump:
        selection_eph_jump:
            telemetry_type: GPS_STATUS
            eph_change_ratio: '>3'
            eph_current: '<1.0'
    condition: position_jump or imu_divergence or satellite_anomaly or eph_jump
    timeframe: 30s
    falsepositives:
        - Expected position jump during hot start
        - Normal satellite constellation change
    level: high
    evidence_strength: high

#!/bin/bash
set -euo pipefail
LOG_FILE=${1:-"/var/log/mavlink/capture.log"}
OUTPUT_DIR="/forensics/mavlink/$(date +%Y%m%d_%H%M%S)"
SENSITIVE_CMDS="176,178,112,241,310,400"
KNOWN_SYSTEMS_FILE="/etc/uas/known_systems.json"
ALERT_THRESHOLD=5

mkdir -p "$OUTPUT_DIR"

parse_mavlink_messages() {
    local logfile=$1
    local output=$2
    python3 - <<'PYEOF' "$logfile" "$output"
import sys, struct, json
from collections import defaultdict

MAVLINK_V2_HEADER = 0xFD
SENSITIVE_CMDS = {176: "ARM", 178: "DISARM", 112: "CALIBRATION",
                  241: "INJECT_FAILURE", 310: "PAYLOAD_DEPLOY", 400: "USER_1"}

def parse_messages(filepath):
    messages = []
    with open(filepath, "rb") as f:
        data = f.read()
    offset = 0
    while offset < len(data) - 10:
        if data[offset] == MAVLINK_V2_HEADER:
            length = data[offset + 1]
            sysid = data[offset + 5]
            compid = data[offset + 6]
            msgid = (data[offset + 9] << 16) | (data[offset + 8] << 8) | data[offset + 7]
            messages.append({
                "sysid": sysid, "compid": compid, "msgid": msgid,
                "cmd_name": MAVLINK_V2_HEADER and MAVLINK_V2_HEADER
            })
            offset += 10 + length + 2
        else:
            offset += 1
    return messages

logfile = sys.argv[1]
output = sys.argv[2]
msgs = parse_messages(logfile)

alerts = {"sensitive_commands": [], "unknown_systems": [], "command_frequency": defaultdict(int)}
known_systems = set()
for msg in msgs:
    sys_key = f"{msg['sysid']}:{msg['compid']}"
    if msg["msgid"] in SENSITIVE_CMDS:
        alerts["sensitive_commands"].append({
            "sysid": msg["sysid"], "msgid": msg["msgid"],
            "cmd": SENSITIVE_CMDS[msg["msgid"]]
        })
    alerts["command_frequency"][msg["msgid"]] += 1
    if sys_key not in known_systems:
        alerts["unknown_systems"].append(sys_key)

with open(f"{output}/mavlink_analysis.json", "w") as f:
    json.dump(alerts, f, indent=2, default=str)
PYEOF
}

echo "[+] MAVLink Forensic Analysis"
echo "  Log file: $LOG_FILE"
echo "  Output: $OUTPUT_DIR"

parse_mavlink_messages "$LOG_FILE" "$OUTPUT_DIR"

echo "[+] Sensitive command analysis"
if [ -f "$OUTPUT_DIR/mavlink_analysis.json" ]; then
    sensitive_count=$(python3 -c "
import json
with open('$OUTPUT_DIR/mavlink_analysis.json') as f:
    d = json.load(f)
print(len(d.get('sensitive_commands', [])))
")
    echo "  Sensitive commands found: $sensitive_count"
    if [ "$sensitive_count" -gt "$ALERT_THRESHOLD" ]; then
        echo "[!] ALERT: High volume of sensitive commands detected"
        python3 -c "
import json
with open('$OUTPUT_DIR/mavlink_analysis.json') as f:
    d = json.load(f)
for cmd in d['sensitive_commands']:
    print(f\"  System {cmd['sysid']}: {cmd['cmd']} (msgid={cmd['msgid']})\")
"
    fi
fi

echo "[+] Unknown system identification"
python3 -c "
import json
with open('$OUTPUT_DIR/mavlink_analysis.json') as f:
    d = json.load(f)
unknowns = set(d.get('unknown_systems', []))
print(f'  Unique systems: {len(unknowns)}')
for s in unknowns:
    print(f'  - System: {s}')
" 2>/dev/null || echo "  Analysis unavailable"

echo "[+] Report saved to $OUTPUT_DIR"

#!/usr/bin/env python3
import hashlib
import json
import os
import sys
from pathlib import Path
from datetime import datetime

class FirmwareHashVerifier:
    def __init__(self, firmware_dir, reference_db_path=None):
        self.firmware_dir = Path(firmware_dir)
        self.reference_db = {}
        self.results = {"verified": [], "mismatched": [], "unknown": [], "errors": []}
        if reference_db_path:
            with open(reference_db_path) as f:
                self.reference_db = json.load(f)

    def calculate_file_hashes(self, filepath):
        hashes = {"sha256": "", "sha1": "", "md5": "", "size": 0}
        sha256 = hashlib.sha256()
        sha1 = hashlib.sha1()
        md5 = hashlib.md5()
        with open(filepath, "rb") as f:
            while True:
                chunk = f.read(8192)
                if not chunk:
                    break
                sha256.update(chunk)
                sha1.update(chunk)
                md5.update(chunk)
        hashes["sha256"] = sha256.hexdigest()
        hashes["sha1"] = sha1.hexdigest()
        hashes["md5"] = md5.hexdigest()
        hashes["size"] = filepath.stat().st_size
        return hashes

    def verify_all(self):
        for root, dirs, files in os.walk(self.firmware_dir):
            for fname in files:
                fpath = Path(root) / fname
                if fpath.suffix.lower() in [".bin", ".hex", ".elf", ".img", ".fw"]:
                    try:
                        hashes = self.calculate_file_hashes(fpath)
                        rel_path = str(fpath.relative_to(self.firmware_dir))
                        if rel_path in self.reference_db:
                            ref = self.reference_db[rel_path]
                            match = all(hashes[k] == ref.get(k, "") for k in ["sha256", "sha1", "md5"])
                            entry = {"file": rel_path, "hashes": hashes, "match": match}
                            if match:
                                self.results["verified"].append(entry)
                            else:
                                self.results["mismatched"].append(entry)
                        else:
                            self.results["unknown"].append({"file": rel_path, "hashes": hashes})
                    except Exception as e:
                        self.results["errors"].append({"file": str(fpath), "error": str(e)})

    def generate_report(self):
        report = {
            "scan_time": datetime.now().isoformat(),
            "firmware_dir": str(self.firmware_dir),
            "summary": {
                "verified": len(self.results["verified"]),
                "mismatched": len(self.results["mismatched"]),
                "unknown": len(self.results["unknown"]),
                "errors": len(self.results["errors"])
            },
            "details": self.results
        }
        return report

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <firmware_directory> [reference_hashes.json]")
        sys.exit(1)
    ref_db = sys.argv[2] if len(sys.argv) > 2 else None
    verifier = FirmwareHashVerifier(sys.argv[1], ref_db)
    verifier.verify_all()
    report = verifier.generate_report()
    print(json.dumps(report, indent=2, ensure_ascii=False))
    if report["summary"]["mismatched"] > 0:
        sys.exit(2)
    sys.exit(0)

rule UAV_Backdoor_Shell {
    meta:
        description = "Detects potential backdoor shell code in UAV firmware"
        author = "UAS Security Forensics"
        date = "2026-07-31"
        reference = "UAS Forensics Framework"
        severity = "critical"
        mitre_attack = "T1059 Command and Scripting Interpreter"
    strings:
        $shell1 = "/bin/sh" ascii
        $shell2 = "/bin/bash" ascii
        $shell3 = "nc -l" ascii nocase
        $shell4 = "ncat" ascii
        $shell5 = "reverse_shell" ascii nocase
        $socket = "socket(AF_INET" ascii
        $connect = "connect(" ascii
        $exec1 = "exec(" ascii
        $exec2 = "system(" ascii
        $base64 = "base64 -d" ascii
        $curl = "curl http" ascii
        $wget = "wget http" ascii
    condition:
        uint32(0) == 0x464C457F and
        ($shell1 or $shell2) and
        ($socket or $connect) and
        ($exec1 or $exec2 or $base64) and
        ($curl or $wget)
}

rule UAV_Hardcoded_Credentials {
    meta:
        description = "Detects hardcoded credentials in UAV firmware"
        author = "UAS Security Forensics"
        date = "2026-07-31"
        severity = "high"
        mitre_attack = "T1552.001 Credentials In Files"
    strings:
        $pw1 = "password=" ascii
        $pw2 = "passwd" ascii
        $api1 = "api_key=" ascii
        $api2 = "apikey" ascii
        $token = "access_token=" ascii
        $secret = "secret_key=" ascii
        $root_pw = "root:0:" ascii
        $root_pw2 = "root:$" ascii
    condition:
        uint32(0) == 0x464C457F and
        3 of ($pw*, $api*, $token, $secret, $root_pw*)
}

rule UAV_GNSS_Spoofing_Module {
    meta:
        description = "Detects GNSS spoofing modules in firmware images"
        author = "UAS Security Forensics"
        date = "2026-07-31"
        severity = "critical"
        mitre_attack = "T1562.001 Impair Defenses"
    strings:
        $gps_spoof = "gps_spoof" ascii nocase
        $gnss_sim = "gnss_sim" ascii nocase
        $nav_hijack = "navigation_hijack" ascii nocase
        $fake_nmea = "$GPGGA" ascii
        $sd_r = "hackrf" ascii nocase
        $iq_write = "iq_write" ascii nocase
        $signal_gen = "signal_generator" ascii nocase
    condition:
        uint32(0) == 0x464C457F and
        2 of ($gps_spoof, $gnss_sim, $nav_hijack) or
        ($fake_nmea and ($sd_r or $iq_write or $signal_gen))
}

#!/usr/bin/env python3
import json
from datetime import datetime

attack_chain = {
    "case_id": "DJI-SEC-2024-001",
    "title": "DJI AeroScope API Unauthorized Access",
    "timeline": [
        {
            "phase": "Reconnaissance",
            "time": "T+0h",
            "action": "Network scanning for AeroScope management port (8080)",
            "technique": "T1046 Network Service Scanning"
        },
        {
            "phase": "Initial Access",
            "time": "T+0h5m",
            "action": "Access AeroScope web management interface without authentication",
            "technique": "T1190 Exploit Public-Facing Application"
        },
        {
            "phase": "Collection",
            "time": "T+0h10m",
            "action": "Extract detected drone serial numbers and flight history",
            "technique": "T1005 Data from Local System"
        },
        {
            "phase": "Exfiltration",
            "time": "T+0h15m",
            "action": "Export detection data to attacker-controlled server",
            "technique": "T1041 Exfiltration Over C2 Channel"
        }
    ],
    "forensic_findings": {
        "access_logs": "No authentication required for API endpoints",
        "data_exposed": ["drone_serial", "flight_timestamp", "operator_location", "detection_radius"],
        "remediation": "Enable authentication, restrict network access, encrypt API traffic"
    }
}

print(json.dumps(attack_chain, indent=2, ensure_ascii=False))