"""
CVE-2025-36911 check

DESCRIPTION:
    This tool tests Bluetooth audio devices for CVE-2025-36911, which allows
    unauthenticated L2CAP connections to audio profiles (A2DP/AVRCP) without
    prior pairing. Vulnerable devices accept connections on restricted PSMs,
    potentially allowing unauthorized audio control or stream access.

AFFECTED PROFILES:
    - AVDTP (PSM 0x0019): Advanced Audio Distribution Profile
    - AVCTP (PSM 0x0017): Audio/Video Remote Control Profile
    - Fast Pair (BLE UUID 0xFE2C): Google Fast Pair protocol

DISCLAIMER:
    Use only on your own devices or with explicit authorization.
    Do not deny pairing requests during testing to avoid implicit bonding.
"""

import asyncio
import socket
import sys
import subprocess
from typing import Dict, List, Optional

from bleak import BleakScanner, BleakClient
from bleak.exc import BleakError

try:
    from bleak.exc import BleakDBusError
except Exception:
    BleakDBusError = BleakError

PSM_AVCTP = 0x0017
PSM_AVDTP = 0x0019
PSM_AVCTP_BROWSING = 0x001B


async def scan_ble_devices() -> List[Dict[str, str]]:
    try:
        devices = await BleakScanner.discover()
        return [{"address": d.address, "name": d.name or ""} for d in devices]
    except BleakDBusError as e:
        print("BLE unavailable (BlueZ/DBus):", e)
        return []
    except BleakError as e:
        print("BLE error:", e)
        return []


async def has_fast_pair_service(addr: str) -> bool:
    if not addr or addr.lower() == "00:00:00:00:00:00":
        return False
    try:
        client = BleakClient(addr)
        await asyncio.wait_for(client.connect(), timeout=8.0)
        try:
            services = client.services
            for s in services:
                if s.uuid.lower() == "0000fe2c-0000-1000-8000-00805f9b34fb":
                    return True
        finally:
            try:
                await client.disconnect()
            except Exception:
                pass
    except (BleakError, asyncio.TimeoutError, asyncio.CancelledError):
        return False
    return False


def l2cap_connect(addr: str, psm: int, timeout: float = 5.0) -> bool:
    sock = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_SEQPACKET, socket.BTPROTO_L2CAP)
    sock.settimeout(timeout)
    try:
        sock.connect((addr, psm))
        return True
    except Exception:
        return False
    finally:
        try:
            sock.close()
        except Exception:
            pass


def test_unpaired_br_edr(addr: str) -> Dict[str, bool]:
    results = {
        "avdtp": l2cap_connect(addr, PSM_AVDTP),
        "avctp": l2cap_connect(addr, PSM_AVCTP),
        "avctp_browsing": l2cap_connect(addr, PSM_AVCTP_BROWSING),
    }
    return results


async def run_tests(target_addr: str) -> None:
    print(f"Target: {target_addr}")
    br_results = test_unpaired_br_edr(target_addr)
    print("\nBR/EDR Audio PSM Results (unauthenticated):")
    for k, v in br_results.items():
        status = "OPEN" if v else "BLOCKED"
        print(f"  {k:20} : {status}")

    fp = await has_fast_pair_service(target_addr)
    print(f"\nFast Pair Service (BLE): {'DETECTED' if fp else 'Not found'}")

    if br_results.get("avdtp") or br_results.get("avctp") or br_results.get("avctp_browsing"):
        print("\n> CVE-2025-36911 VULNERABILITY DETECTED")
        print("> Device accepts unauthenticated L2CAP connections to audio profiles")
    else:
        print("\n> Device not vulnerable to CVE-2025-36911 - pairing required for audio access")


def bluetoothctl_info(addr: str) -> Dict[str, str]:
    info: Dict[str, str] = {}
    try:
        out = subprocess.check_output(["bluetoothctl", "info", addr], text=True)
        for line in out.splitlines():
            line = line.strip()
            if line.startswith("Paired:"):
                info["paired"] = line.split(":", 1)[1].strip()
            elif line.startswith("Connected:"):
                info["connected"] = line.split(":", 1)[1].strip()
            elif line.startswith("Trusted:"):
                info["trusted"] = line.split(":", 1)[1].strip()
            elif line.startswith("Blocked:"):
                info["blocked"] = line.split(":", 1)[1].strip()
            elif line.startswith("Bonded:"):
                info["bonded"] = line.split(":", 1)[1].strip()
    except Exception:
        pass
    return info


def bluetoothctl_cmd(*args: str) -> bool:
    try:
        subprocess.check_call(["bluetoothctl", *args], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return True
    except Exception:
        return False


def validate_mac_address(addr: str) -> bool:
    if not addr:
        return False
    parts = addr.split(":")
    if len(parts) != 6:
        return False
    for part in parts:
        if len(part) != 2:
            return False
        try:
            int(part, 16)
        except ValueError:
            return False
    return True


def device_exists(addr: str) -> bool:
    try:
        out = subprocess.check_output(["bluetoothctl", "devices"], text=True)
        for line in out.splitlines():
            if addr.lower() in line.lower():
                return True
        return False
    except Exception:
        return False


def print_help() -> None:
    print("""
CVE-2025-36911 scanner

USAGE:
    python CVE-2025-36911.py [OPTIONS] [BLUETOOTH_ADDRESS]

OPTIONS:
    --help                 Display this help message
    --skip-ble             Skip BLE device scanning (BR/EDR only mode, you must provide address)
    --require-unpaired     Refuse to test already paired devices (usefull to avoid false positives)
    --lock-pairing         Disable pairing during test (pairable off, usefull to avoid implicit bonding)

ARGUMENTS:
    BLUETOOTH_ADDRESS     Target device address (XX:XX:XX:XX:XX:XX)
                          If not provided, BLE scan will discover devices (if --skip-ble not used)

EXAMPLES:
    python CVE-2025-36911.py
        Interactive mode: scan BLE devices and select target

    python CVE-2025-36911.py AA:BB:CC:DD:EE:FF
        Test specific device by address

    python CVE-2025-36911.py --require-unpaired --lock-pairing AA:BB:CC:DD:EE:FF
        Test unpaired device with pairing locked during test

DESCRIPTION:
    Tests Bluetooth audio devices for CVE-2025-36911 vulnerability by attempting
    unauthenticated L2CAP connections to audio profiles (AVDTP/AVCTP). Vulnerable
    devices accept connections without prior pairing, allowing unauthorized audio
    control or stream access.

DISCLAIMER:
    Use only on your own devices or with explicit authorization.
""")


async def main(argv: List[str]) -> None:
    """Main entry point with argument parsing."""
    if "--help" in argv or "-h" in argv:
        print_help()
        return

    skip_ble = False
    require_unpaired = False
    lock_pairing = False
    target: Optional[str] = None

    for a in argv[1:]:
        if a == "--skip-ble":
            skip_ble = True
        elif a == "--require-unpaired":
            require_unpaired = True
        elif a == "--lock-pairing":
            lock_pairing = True
        elif ":" in a:
            if validate_mac_address(a):
                target = a
            else:
                print(f"Error: Invalid MAC address format: {a}")
                print("MAC address must be in format XX:XX:XX:XX:XX:XX")
                return

    if not target:
        if not skip_ble:
            print(" Scanning BLE devices...")
            devices = await scan_ble_devices()
            for i, d in enumerate(devices):
                print(f"[{i}] {d['address']} - {d['name']}")
            if devices:
                sel = input("\nSelect target index or enter BR/EDR address: ").strip()
                if ":" in sel:
                    if validate_mac_address(sel):
                        target = sel
                    else:
                        print(f"Error: Invalid MAC address format: {sel}")
                        print("MAC address must be in format XX:XX:XX:XX:XX:XX")
                        return
                else:
                    try:
                        idx = int(sel)
                        target = devices[idx]["address"]
                    except Exception:
                        print("Invalid selection.")
                        return
            else:
                print("No BLE devices found. Have you turn on Bluetooth?")
                return
        else:
            print("Provide BR/EDR address as argument.")
            return

    if target and require_unpaired:
        info = bluetoothctl_info(target)
        paired = info.get("paired")
        if paired is None:
            print(f"Warning: Cannot determine pairing state for {target}")
            print("The device may not be currently discoverable.")
            print("\nOptions:")
            print("  1. Remove --skip-ble to scan and discover the device first")
            print("  2. Remove --require-unpaired to test regardless of pairing state")
            print("  3. Ensure the device is powered on and in pairing mode")
            return
        if paired.lower() == "yes":
            print("Error: Device already paired. Unpair before testing.")
            return

    if target:
        if not device_exists(target):
            print(f"Error: Device {target} not found in Bluetooth cache.")
            print("Please ensure the device is discoverable and has been scanned.")
            print("Run without address to scan for available devices first.")
            return

    if lock_pairing:
        bluetoothctl_cmd("pairable", "off")

    await run_tests(target)

    if target:
        info_after = bluetoothctl_info(target)
        if info_after.get("paired", "no").lower() == "yes":
            print("\n> WARNING: Device is now paired.")
            print("> L2CAP connections may have triggered implicit bonding.")

    if lock_pairing:
        bluetoothctl_cmd("pairable", "on")


if __name__ == "__main__":
    try:
        asyncio.run(main(sys.argv))
    except KeyboardInterrupt:
        print("\n Interrupted.")