BLE GATT Fuzz

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
BLE GATT Fuzzer — mutates writes against writable characteristics on a live BLE session.
"""

from __future__ import annotations

import json
import random
import time
from datetime import datetime
from typing import Dict, List, Optional, Tuple

from kittysploit import *
from lib.protocols.ble.ble_client import normalize_uuid
from lib.protocols.ble.ble_session_mixin import BleSessionMixin


STRATEGIES = (
    "random",
    "overflow",
    "nulls",
    "ff",
    "walking",
    "ascii",
    "mixed",
)


class Module(Post, BleSessionMixin):
    __info__ = {
        "name": "BLE GATT Fuzz",
        "description": (
            "Fuzzes writable GATT characteristics with mutated payloads to probe "
            "firmware input handling (disconnects, errors, anomalous notifies)"
        ),
        "author": "KittySploit Team",
        "version": "1.0.0",
        "session_type": SessionType.BLE,
        "tags": ["ble", "bluetooth", "iot", "fuzz", "gatt", "exploit"],
        "references": [
            "https://www.bluetooth.com/specifications/gatt/",
            "https://attack.mitre.org/techniques/T0866/",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation", "wireless_probe"],
            "expected_requests": 50,
            "reversible": False,
            "approval_required": True,
            "produces": ["risk_signals"],
            "cost": 1.5,
            "noise": 0.7,
            "value": 1.3,
            "chain": {
                "consumes_capabilities": ["ble_characteristics", "ble_session"],
                "produces_capabilities": [
                    {"capability": "ble_fuzz_results", "from_detail": ""},
                ],
                "suggested_followups": [
                    "post/ble/gather/notify_capture",
                    "post/ble/manage/read_characteristic",
                ],
            },
        },
    }

    uuid = OptString(
        "",
        "Target characteristic UUID (empty = all writable characteristics)",
        False,
    )
    strategy = OptChoice(
        "mixed",
        "Mutation strategy",
        True,
        choices=list(STRATEGIES),
    )
    iterations = OptInteger(50, "Payloads per characteristic", True)
    delay = OptFloat(0.05, "Delay between writes (seconds)", True)
    min_len = OptInteger(0, "Minimum payload length", False)
    max_len = OptInteger(64, "Maximum payload length (ATT payloads are often <= MTU-3)", False)
    with_response = OptChoice(
        "auto",
        "Write mode: auto / true (write) / false (write-without-response)",
        False,
        choices=["auto", "true", "false"],
    )
    stop_on_disconnect = OptBool(True, "Stop fuzzing if the GATT link drops", True)
    subscribe_notify = OptBool(
        False,
        "Subscribe to notify/indicate on target chars while fuzzing",
        False,
    )
    dry_run = OptBool(False, "List targets and sample payloads — do not write", False)
    seed = OptInteger(0, "PRNG seed (0 = time-based)", False, advanced=True)
    export_json = OptString("", "Optional JSON report path", False)
    store_results = OptBool(True, "Store report in session.data['ble_gatt_fuzz']", False)

    def check(self):
        sid = str(self.session_id or "").strip()
        if not sid:
            print_error("Session ID not set")
            return False
        session = self.framework.session_manager.get_session(sid) if self.framework else None
        if not session:
            print_error("Session not found")
            return False
        if str(session.session_type).lower() != SessionType.BLE.value:
            print_error(f"Session is not BLE (type: {session.session_type})")
            return False
        if int(self.iterations or 0) <= 0:
            print_error("iterations must be > 0")
            return False
        if int(self.min_len or 0) < 0 or int(self.max_len or 0) < int(self.min_len or 0):
            print_error("invalid min_len / max_len")
            return False
        try:
            self.open_ble()
            return True
        except Exception as exc:
            print_error(str(exc))
            return False

    def run(self):
        client = self.open_ble()
        info = client.connection_summary()
        seed = int(self.seed or 0)
        if seed <= 0:
            seed = int(time.time()) & 0xFFFFFFFF
        rng = random.Random(seed)

        targets = self._select_targets(client)
        if not targets:
            print_error("No writable GATT characteristics found")
            return False

        iterations = max(1, int(self.iterations or 50))
        delay = max(0.0, float(self.delay or 0))
        strategy = str(self.strategy or "mixed")
        min_len = max(0, int(self.min_len or 0))
        max_len = max(min_len, int(self.max_len or 64))

        print_info("=" * 80)
        print_success("BLE GATT Fuzz")
        print_info(f"  device     : {info.get('address')}")
        print_info(f"  strategy   : {strategy}")
        print_info(f"  targets    : {len(targets)}")
        print_info(f"  iterations : {iterations} / char")
        print_info(f"  lengths    : {min_len}-{max_len}")
        print_info(f"  seed       : {seed}")
        print_warning("Fuzzing may crash or soft-brick poorly written firmware")
        print_info("=" * 80)

        for char in targets:
            props = ",".join(char["properties"]) or "-"
            print_info(f"  target {char['uuid']}  [{props}]  service={char['service_uuid']}")

        if bool(self.dry_run):
            samples = [
                self._mutate(strategy, rng, min_len, max_len).hex() or "(empty)"
                for _ in range(min(5, iterations))
            ]
            print_success("Dry run — sample payloads: " + ", ".join(samples))
            return True

        subscribed = []
        if bool(self.subscribe_notify):
            for char in targets:
                props = set(p.lower() for p in char["properties"])
                if "notify" in props or "indicate" in props:
                    try:
                        client.start_notify(char["uuid"])
                        subscribed.append(char["uuid"])
                    except Exception as exc:
                        print_warning(f"notify subscribe failed for {char['uuid']}: {exc}")

        report = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "address": info.get("address"),
            "strategy": strategy,
            "seed": seed,
            "iterations_per_char": iterations,
            "min_len": min_len,
            "max_len": max_len,
            "targets": [t["uuid"] for t in targets],
            "writes_ok": 0,
            "writes_error": 0,
            "disconnect": False,
            "anomalies": [],
            "errors": [],
            "notify_events": 0,
        }

        try:
            for char in targets:
                uuid = char["uuid"]
                response = self._resolve_response(char)
                print_status(f"Fuzzing {uuid} ({iterations} payloads, response={response})")
                for i in range(iterations):
                    if bool(self.stop_on_disconnect) and not client.connected:
                        report["disconnect"] = True
                        report["anomalies"].append(
                            {
                                "type": "disconnect",
                                "uuid": uuid,
                                "iteration": i,
                                "detail": "GATT link dropped during fuzzing",
                            }
                        )
                        print_error(f"Disconnected while fuzzing {uuid} at iteration {i}")
                        break

                    payload = self._mutate(strategy, rng, min_len, max_len)
                    try:
                        client.write_characteristic(uuid, payload, response=response)
                        report["writes_ok"] += 1
                    except Exception as exc:
                        report["writes_error"] += 1
                        err = {
                            "uuid": uuid,
                            "iteration": i,
                            "payload_hex": payload.hex(),
                            "payload_len": len(payload),
                            "error": str(exc),
                        }
                        report["errors"].append(err)
                        # Keep going unless link is dead
                        if not client.connected:
                            report["disconnect"] = True
                            report["anomalies"].append(
                                {
                                    "type": "disconnect_after_error",
                                    "uuid": uuid,
                                    "iteration": i,
                                    "payload_hex": payload.hex(),
                                    "error": str(exc),
                                }
                            )
                            print_error(f"Write error + disconnect on {uuid}: {exc}")
                            break
                        if i < 5 or i % 25 == 0:
                            print_warning(f"  [{i}] write error: {exc}")

                    if delay:
                        time.sleep(delay)

                if report["disconnect"] and bool(self.stop_on_disconnect):
                    break

            # Drain any notifications captured during the run
            try:
                events = client.drain_notifications()
                report["notify_events"] = len(events)
                if events:
                    report["notify_sample"] = [
                        {"uuid": e.uuid, "hex": e.hex, "len": len(e.data)}
                        for e in events[:20]
                    ]
            except Exception:
                pass
        finally:
            for uuid in subscribed:
                try:
                    client.stop_notify(uuid)
                except Exception:
                    pass

        print_info("-" * 80)
        print_success(
            f"Done — ok={report['writes_ok']} errors={report['writes_error']} "
            f"disconnect={report['disconnect']} notifies={report['notify_events']}"
        )
        if report["anomalies"]:
            print_warning(f"Anomalies: {len(report['anomalies'])}")
            for item in report["anomalies"][:10]:
                print_warning(f"  {item}")

        if bool(self.store_results):
            session = self._resolve_session()
            if session:
                data = session.data if isinstance(session.data, dict) else {}
                data["ble_gatt_fuzz"] = report
                session.data = data

        out = str(self.export_json or "").strip()
        if out:
            try:
                with open(out, "w", encoding="utf-8") as handle:
                    json.dump(report, handle, indent=2)
                print_success(f"Report written to {out}")
            except OSError as exc:
                print_error(f"Failed to write report: {exc}")

        return True

    def _select_targets(self, client) -> List[Dict]:
        wanted = str(self.uuid or "").strip()
        wanted_norm = normalize_uuid(wanted) if wanted else ""
        services = client.get_services(refresh=False)
        targets = []
        for svc in services:
            for char in svc.characteristics:
                props = [p.lower() for p in char.properties]
                writable = "write" in props or "write-without-response" in props
                if not writable:
                    continue
                if wanted_norm and normalize_uuid(char.uuid) != wanted_norm and char.uuid.lower() != wanted.lower():
                    continue
                targets.append(
                    {
                        "uuid": char.uuid,
                        "service_uuid": svc.uuid,
                        "properties": char.properties,
                        "handle": char.handle,
                    }
                )
        return targets

    def _resolve_response(self, char: Dict) -> Optional[bool]:
        mode = str(self.with_response or "auto").lower()
        if mode == "true":
            return True
        if mode == "false":
            return False
        props = set(p.lower() for p in char.get("properties") or [])
        if "write" in props:
            return True
        if "write-without-response" in props:
            return False
        return None

    def _mutate(self, strategy: str, rng: random.Random, min_len: int, max_len: int) -> bytes:
        strategy = (strategy or "mixed").lower()
        if strategy == "mixed":
            strategy = rng.choice(["random", "overflow", "nulls", "ff", "walking", "ascii"])

        if strategy == "overflow":
            # Bias toward boundary lengths
            candidates = sorted(
                {
                    min_len,
                    max_len,
                    min(max_len, 20),
                    min(max_len, 23),
                    min(max_len, 244),
                    min(max_len, 512),
                    max(min_len, max_len // 2),
                }
            )
            length = rng.choice([c for c in candidates if min_len <= c <= max_len] or [max_len])
            pattern = rng.choice([0x00, 0x41, 0xFF, 0x20])
            return bytes([pattern]) * length

        if strategy == "nulls":
            length = rng.randint(min_len, max_len)
            return b"\x00" * length

        if strategy == "ff":
            length = rng.randint(min_len, max_len)
            return b"\xff" * length

        if strategy == "walking":
            length = max(1, rng.randint(max(1, min_len), max(1, max_len)))
            buf = bytearray(length)
            idx = rng.randint(0, length - 1)
            buf[idx] = rng.choice([0x01, 0x02, 0x80, 0xFF])
            return bytes(buf)

        if strategy == "ascii":
            length = rng.randint(min_len, max_len)
            if length == 0:
                return b""
            alphabet = (
                "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
                "%s%x%n../\\\"';{}[]<>\x00"
            )
            # Keep payload bytes valid; map weird markers to printable-ish
            out = bytearray()
            for _ in range(length):
                ch = rng.choice(alphabet)
                out.append(ord(ch) if ord(ch) < 256 else 0x3F)
            return bytes(out)

        # random
        length = rng.randint(min_len, max_len)
        return bytes(rng.getrandbits(8) for _ in range(length))