Add PolterHID awareness training firmware

README.md

@@ -0,0 +1,43 @@
+# PolterHID
+
+A deliberately limited MicroPython awareness-training firmware for an ESP32 USB stick. On boot, it presents as a USB keyboard/mouse and, at independently random intervals, sends only:
+
+- one `Return` key press and release; and
+- a one-pixel relative mouse movement.
+
+The authenticated web page also provides manual **Volume up** and **Volume down** controls for a clearly observable, non-destructive HID demonstration.
+
+A password-protected local web page changes the enabled event types and their timing. The device connects to one preconfigured Wi-Fi network; it does not create an access point.
+
+## Hardware and firmware requirement
+
+This needs an **ESP32-S2 or ESP32-S3** board/stick whose USB connector is wired to the chip's native USB peripheral. A classic ESP32 with a CP210x/CH340 USB-to-serial adapter cannot act as a USB HID device.
+
+Flash a current MicroPython build for that exact board which provides `machine.USBDevice`. Check at the REPL:
+
+```python
+from machine import USBDevice
+```
+
+If that import fails, use an up-to-date native-USB MicroPython build for the board. This project uses the custom USB-device API and does not work with CircuitPython's `usb_hid` API.
+
+## Install
+
+1. Copy `settings.example.json` to the device filesystem as `settings.json` and replace all three credentials. Do this **before first boot**: the firmware cannot join Wi-Fi with the placeholder defaults.
+2. Copy `main.py`, `hid.py`, `settings.py`, and `web.py` to the device root filesystem.
+3. Reset the board and plug its native USB connector into the demonstration host.
+4. Once it joins Wi-Fi, visit `http://polterhid.local/`. If the client/network does not support mDNS, find the device IP address in the training Wi-Fi DHCP leases and visit `http://DEVICE_IP/` instead.
+5. Sign in with username `admin` and the `web_password` from `settings.json`.
+
+`main.py` runs automatically after boot in MicroPython. It advertises the HTTP page through mDNS as `polterhid.local`; this requires a MicroPython ESP32 build containing the `mdns` module. Configuration saved in the web page is retained in `settings.json` and takes effect immediately.
+
+## Operational safeguards
+
+- Change the example credentials before deployment. HTTP Basic authentication and all form data, including the password, are unencrypted because this intentionally has no TLS. Use an isolated/trusted training Wi-Fi network.
+- The page provides an immediate **Enable all injection** switch. Turn it off before handing the device to anyone or ending a session.
+- This firmware intentionally has no shell commands, keystroke payloads, text injection, storage emulation, Wi-Fi scanning, AP mode, or remote firmware-update endpoint.
+- Test first on a dedicated demonstration machine. The configured defaults are intentionally infrequent but still cause a Return and minimal mouse movement.
+
+## USB implementation note
+
+`hid.py` exposes one composite HID interface with keyboard report ID `1`, mouse report ID `2`, and Consumer Control/media report ID `3` (Volume up/down). It queues reports while the endpoint is busy and discards a transfer if the host disconnects, avoiding stale events being replayed after a reconnect.

hid.py

@@ -0,0 +1,137 @@
+"""Minimal composite keyboard/mouse USB HID device for MicroPython ESP32-S2/S3.
+
+Requires a MicroPython build exposing `machine.USBDevice` (native USB device
+support). The report descriptor uses report ID 1 for a boot-style keyboard,
+report ID 2 for a relative three-button mouse, and report ID 3 for media keys.
+"""
+
+from machine import USBDevice
+
+# Combined keyboard, mouse, and Consumer Control HID report descriptor.
+REPORT_DESCRIPTOR = bytes((
+    0x05, 0x01, 0x09, 0x06, 0xA1, 0x01, 0x85, 0x01,
+    0x05, 0x07, 0x19, 0xE0, 0x29, 0xE7, 0x15, 0x00, 0x25, 0x01,
+    0x75, 0x01, 0x95, 0x08, 0x81, 0x02, 0x95, 0x01, 0x75, 0x08,
+    0x81, 0x01, 0x95, 0x06, 0x75, 0x08, 0x15, 0x00, 0x25, 0x65,
+    0x05, 0x07, 0x19, 0x00, 0x29, 0x65, 0x81, 0x00, 0xC0,
+    0x05, 0x01, 0x09, 0x02, 0xA1, 0x01, 0x85, 0x02, 0x09, 0x01,
+    0xA1, 0x00, 0x05, 0x09, 0x19, 0x01, 0x29, 0x03, 0x15, 0x00,
+    0x25, 0x01, 0x95, 0x03, 0x75, 0x01, 0x81, 0x02, 0x95, 0x01,
+    0x75, 0x05, 0x81, 0x01, 0x05, 0x01, 0x09, 0x30, 0x09, 0x31,
+    0x15, 0x81, 0x25, 0x7F, 0x75, 0x08, 0x95, 0x02, 0x81, 0x06,
+    0xC0, 0xC0,
+    # Consumer Control: Volume Increment and Volume Decrement.
+    0x05, 0x0C, 0x09, 0x01, 0xA1, 0x01, 0x85, 0x03,
+    0x15, 0x00, 0x25, 0x01, 0x09, 0xE9, 0x09, 0xEA,
+    0x75, 0x01, 0x95, 0x02, 0x81, 0x02,
+    0x75, 0x06, 0x95, 0x01, 0x81, 0x01, 0xC0,
+))
+
+# Configuration descriptor: one HID interface, one interrupt-IN endpoint.
+CONFIG_DESCRIPTOR = bytes((
+    0x09, 0x02, 0x22, 0x00, 0x01, 0x01, 0x00, 0x80, 0x32,
+    0x09, 0x04, 0x00, 0x00, 0x01, 0x03, 0x00, 0x00, 0x00,
+    0x09, 0x21, 0x11, 0x01, 0x00, 0x01, 0x22,
+    len(REPORT_DESCRIPTOR) & 0xFF, len(REPORT_DESCRIPTOR) >> 8,
+    0x07, 0x05, 0x81, 0x03, 0x10, 0x00, 0x0A,
+))
+
+# HID class requests used by common desktop hosts.
+_GET_DESCRIPTOR = 0x06
+_DESCRIPTOR_TYPE_REPORT = 0x22
+
+
+class HID:
+    """Queues one USB report at a time, avoiding writes to a busy endpoint."""
+
+    def __init__(self):
+        self._usb = USBDevice()
+        self._ready = False
+        self._busy = False
+        self._pending = []
+        self._configure()
+
+    def _configure(self):
+        # USBDevice's custom-device API was introduced with a positional
+        # BUILTIN_NONE argument. The fallback supports earlier preview builds.
+        kwargs = {
+            "strs": [None, "PolterHID", "Awareness trainer", "PHID-001"],
+            "open_itf_cb": self._open_interface,
+            "reset_cb": self._reset,
+            "control_xfer_cb": self._control_transfer,
+            "xfer_cb": self._transfer_done,
+        }
+        builtin_none = getattr(USBDevice, "BUILTIN_NONE", None)
+        if builtin_none is None:
+            # Current stable custom-device API.
+            self._usb.config(CONFIG_DESCRIPTOR, **kwargs)
+        else:
+            # API variant that requires an explicit built-in driver selection.
+            self._usb.config(builtin_none, CONFIG_DESCRIPTOR, **kwargs)
+
+    def _open_interface(self, interface_descriptor):
+        # Endpoint descriptors follow the nine-byte interface descriptor.
+        offset = 9
+        while offset + 1 < len(interface_descriptor):
+            length = interface_descriptor[offset]
+            descriptor_type = interface_descriptor[offset + 1]
+            if length < 2:
+                break
+            if descriptor_type == 0x05:
+                self._usb.ep_open(interface_descriptor[offset:offset + length])
+            offset += length
+        self._ready = True
+
+    def _reset(self):
+        self._ready = False
+        self._busy = False
+        self._pending = []
+
+    def _control_transfer(self, stage, request):
+        # Return our report descriptor during the setup/data stage. HID has no
+        # feature reports or output reports in this intentionally small design.
+        if stage == 1 and request[1] == _GET_DESCRIPTOR and request[3] == _DESCRIPTOR_TYPE_REPORT:
+            return REPORT_DESCRIPTOR
+        return False
+
+    def _transfer_done(self, endpoint, result, transferred):
+        if endpoint == 0x81:
+            self._busy = False
+            self._flush()
+
+    def _flush(self):
+        if not self._ready or self._busy or not self._pending:
+            return
+        report = self._pending.pop(0)
+        try:
+            self._usb.submit_xfer(0x81, report)
+            self._busy = True
+        except OSError:
+            # A host may disconnect between scheduling and transfer. Dropping
+            # an old movement/key event is safer than replaying it on reconnect.
+            self._busy = False
+
+    def _send(self, report):
+        self._pending.append(report)
+        self._flush()
+
+    def press_return(self):
+        # Report ID, modifiers, reserved, six key slots. 0x28 is Enter/Return.
+        self._send(bytes((1, 0, 0, 0x28, 0, 0, 0, 0, 0)))
+
+    def release_keys(self):
+        self._send(bytes((1, 0, 0, 0, 0, 0, 0, 0, 0)))
+
+    def move(self, x, y):
+        x = max(-1, min(1, int(x)))
+        y = max(-1, min(1, int(y)))
+        self._send(bytes((2, 0, x & 0xFF, y & 0xFF, 0)))
+
+    def press_volume_up(self):
+        self._send(bytes((3, 0x01)))
+
+    def press_volume_down(self):
+        self._send(bytes((3, 0x02)))
+
+    def release_consumer(self):
+        self._send(bytes((3, 0)))

main.py

@@ -0,0 +1,125 @@
+"""PolterHID application entry point."""
+
+import network
+import time
+import urandom
+
+from hid import HID
+from settings import load, save
+from web import WebServer
+
+KEY_HOLD_MS = 80
+
+
+def connect_wifi(settings):
+    wlan = network.WLAN(network.STA_IF)
+    wlan.active(True)
+    if not wlan.isconnected():
+        wlan.connect(settings["wifi_ssid"], settings["wifi_password"])
+    return wlan
+
+
+def seconds_to_ms(seconds):
+    return seconds * 1000
+
+
+def randint(minimum, maximum):
+    """Return an inclusive random integer using MicroPython's core RNG API."""
+    return minimum + (urandom.getrandbits(30) % (maximum - minimum + 1))
+
+
+def choose_due(now, minimum, maximum):
+    return time.ticks_add(now, seconds_to_ms(randint(minimum, maximum)))
+
+
+def start_mdns():
+    """Advertise the configuration page as http://polterhid.local/."""
+    import mdns
+
+    server = mdns.Server()
+    server.init("polterhid", "PolterHID")
+    server.advertise_service("_http", "_tcp", 80, "PolterHID")
+    return server
+
+
+def main():
+    settings = load()
+    hid = HID()
+    wlan = connect_wifi(settings)
+    volume_requests = []
+
+    def request_volume(action):
+        # Keep the UI responsive but bound queued manual actions.
+        if len(volume_requests) < 8:
+            volume_requests.append(action)
+
+    server = WebServer(settings, lambda: save(settings), request_volume)
+
+    now = time.ticks_ms()
+    next_return = choose_due(now, settings["return_min_seconds"], settings["return_max_seconds"])
+    next_mouse = choose_due(now, settings["mouse_min_seconds"], settings["mouse_max_seconds"])
+    release_at = None
+    consumer_release_at = None
+    next_wifi_retry = now
+    mdns_server = None
+    next_mdns_retry = now
+
+    while True:
+        now = time.ticks_ms()
+        server.poll()
+
+        # Reconnect without blocking injections or the web server indefinitely.
+        if not wlan.isconnected() and time.ticks_diff(now, next_wifi_retry) >= 0:
+            wlan.connect(settings["wifi_ssid"], settings["wifi_password"])
+            next_wifi_retry = time.ticks_add(now, 10000)
+
+        if (wlan.isconnected() and mdns_server is None and
+                time.ticks_diff(now, next_mdns_retry) >= 0):
+            try:
+                mdns_server = start_mdns()
+                print("mDNS available at http://polterhid.local/")
+            except (ImportError, OSError, AttributeError) as error:
+                # Keep HID and web handling available by IP if this firmware
+                # build lacks mDNS or Wi-Fi is not ready yet.
+                print("mDNS unavailable:", error)
+                next_mdns_retry = time.ticks_add(now, 10000)
+
+        if release_at is not None and time.ticks_diff(now, release_at) >= 0:
+            hid.release_keys()
+            release_at = None
+
+        if consumer_release_at is not None and time.ticks_diff(now, consumer_release_at) >= 0:
+            hid.release_consumer()
+            consumer_release_at = None
+        elif consumer_release_at is None and volume_requests:
+            if volume_requests.pop(0) == "volume_up":
+                hid.press_volume_up()
+            else:
+                hid.press_volume_down()
+            consumer_release_at = time.ticks_add(now, KEY_HOLD_MS)
+
+        if (settings["enabled"] and settings["return_enabled"] and
+                release_at is None and time.ticks_diff(now, next_return) >= 0):
+            hid.press_return()
+            release_at = time.ticks_add(now, KEY_HOLD_MS)
+            next_return = choose_due(
+                now, settings["return_min_seconds"], settings["return_max_seconds"]
+            )
+
+        if (settings["enabled"] and settings["mouse_enabled"] and
+                time.ticks_diff(now, next_mouse) >= 0):
+            # Never send (0, 0), so each scheduled jitter is observable.
+            x = randint(-1, 1)
+            y = randint(-1, 1)
+            if x == 0 and y == 0:
+                x = 1
+            hid.move(x, y)
+            next_mouse = choose_due(
+                now, settings["mouse_min_seconds"], settings["mouse_max_seconds"]
+            )
+
+        # Do not busy-loop; 25 ms keeps the 80 ms key hold responsive.
+        time.sleep_ms(25)
+
+
+main()

settings.example.json

@@ -0,0 +1,12 @@
+{
+  "wifi_ssid": "TrainingWiFi",
+  "wifi_password": "replace-with-the-network-password",
+  "web_password": "replace-with-a-long-admin-password",
+  "enabled": true,
+  "return_enabled": true,
+  "mouse_enabled": true,
+  "return_min_seconds": 45,
+  "return_max_seconds": 120,
+  "mouse_min_seconds": 20,
+  "mouse_max_seconds": 60
+}

settings.py

@@ -0,0 +1,90 @@
+"""Persistent configuration for PolterHID.
+
+`settings.json` is deliberately kept outside the source tree and is created on
+first boot. The web password is stored in clear text because TLS is explicitly
+not used; use this only on a trusted training Wi-Fi network.
+"""
+
+try:
+    import ujson as json
+except ImportError:
+    import json
+
+DEFAULTS = {
+    "wifi_ssid": "CHANGE_ME",
+    "wifi_password": "CHANGE_ME",
+    "web_password": "change-this-before-deploying",
+    "enabled": True,
+    "return_enabled": True,
+    "mouse_enabled": True,
+    "return_min_seconds": 45,
+    "return_max_seconds": 120,
+    "mouse_min_seconds": 20,
+    "mouse_max_seconds": 60,
+}
+
+SETTINGS_FILE = "settings.json"
+
+
+def load():
+    """Return validated settings, creating the file on first boot."""
+    values = DEFAULTS.copy()
+    try:
+        with open(SETTINGS_FILE, "r") as file:
+            stored = json.load(file)
+        if isinstance(stored, dict):
+            for key in DEFAULTS:
+                if key in stored:
+                    values[key] = stored[key]
+    except (OSError, ValueError):
+        save(values)
+
+    _validate(values)
+    return values
+
+
+def save(values):
+    """Atomically replace the settings file where the port supports rename."""
+    _validate(values)
+    temporary = SETTINGS_FILE + ".tmp"
+    with open(temporary, "w") as file:
+        json.dump(values, file)
+    try:
+        import os
+        os.remove(SETTINGS_FILE)
+    except OSError:
+        pass
+    import os
+    os.rename(temporary, SETTINGS_FILE)
+
+
+def apply_form(values, form):
+    """Update configurable runtime values from a decoded HTTP form."""
+    for key in ("enabled", "return_enabled", "mouse_enabled"):
+        values[key] = form.get(key) == "on"
+
+    for key in (
+        "return_min_seconds", "return_max_seconds",
+        "mouse_min_seconds", "mouse_max_seconds",
+    ):
+        if key in form:
+            try:
+                values[key] = int(form[key])
+            except (TypeError, ValueError):
+                raise ValueError("%s must be a whole number" % key)
+
+    _validate(values)
+
+
+def _validate(values):
+    for key in ("wifi_ssid", "wifi_password", "web_password"):
+        if not isinstance(values.get(key), str) or not values[key]:
+            raise ValueError("%s must be a non-empty string" % key)
+
+    for prefix in ("return", "mouse"):
+        minimum = values[prefix + "_min_seconds"]
+        maximum = values[prefix + "_max_seconds"]
+        if not isinstance(minimum, int) or not isinstance(maximum, int):
+            raise ValueError("%s intervals must be integers" % prefix)
+        if minimum < 1 or maximum < minimum or maximum > 86400:
+            raise ValueError("invalid %s interval" % prefix)

web.py

@@ -0,0 +1,160 @@
+"""Small non-blocking password-protected HTTP configuration interface."""
+
+import ubinascii
+import socket
+
+
+class WebServer:
+    def __init__(self, settings, on_save, on_volume):
+        self.settings = settings
+        self.on_save = on_save
+        self.on_volume = on_volume
+        self._client = None
+        self._buffer = b""
+        self._socket = socket.socket()
+        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._socket.bind(("0.0.0.0", 80))
+        self._socket.listen(1)
+        self._socket.settimeout(0)
+
+    def poll(self):
+        if self._client is None:
+            try:
+                self._client, _ = self._socket.accept()
+                self._client.settimeout(0)
+                self._buffer = b""
+            except OSError:
+                return
+        try:
+            data = self._client.recv(512)
+            if data:
+                self._buffer += data
+            elif not self._buffer:
+                self._close()
+                return
+        except OSError:
+            pass
+
+        if len(self._buffer) > 4096:
+            self._reply(413, "Request too large")
+            return
+        marker = self._buffer.find(b"\r\n\r\n")
+        if marker < 0:
+            return
+        headers = self._buffer[:marker].decode("utf-8", "replace")
+        length = self._content_length(headers)
+        if len(self._buffer) < marker + 4 + length:
+            return
+        body = self._buffer[marker + 4:marker + 4 + length].decode("utf-8", "replace")
+        self._handle(headers, body)
+
+    def _handle(self, headers, body):
+        lines = headers.split("\r\n")
+        request = lines[0].split(" ")
+        if len(request) < 2:
+            self._reply(400, "Malformed request")
+            return
+        if not self._authorized(lines):
+            self._reply(401, "Authentication required", {"WWW-Authenticate": 'Basic realm="PolterHID"'})
+            return
+
+        if request[0] == "POST" and request[1] == "/":
+            try:
+                from settings import apply_form
+                form = _parse_form(body)
+                apply_form(self.settings, form)
+                action = form.get("action")
+                if action in ("volume_up", "volume_down"):
+                    self.on_volume(action)
+                self.on_save()
+                self._reply(303, "", {"Location": "/"})
+            except ValueError as error:
+                self._reply(400, str(error))
+            return
+        self._reply(200, _page(self.settings), {"Content-Type": "text/html; charset=utf-8"})
+
+    def _authorized(self, lines):
+        expected = "Basic " + ubinascii.b2a_base64(
+            ("admin:" + self.settings["web_password"]).encode()
+        ).strip().decode()
+        for line in lines[1:]:
+            if line.lower().startswith("authorization:"):
+                return line.split(":", 1)[1].strip() == expected
+        return False
+
+    @staticmethod
+    def _content_length(headers):
+        for line in headers.split("\r\n")[1:]:
+            if line.lower().startswith("content-length:"):
+                try:
+                    return int(line.split(":", 1)[1].strip())
+                except ValueError:
+                    return 0
+        return 0
+
+    def _reply(self, status, body, extra=None):
+        reasons = {200: "OK", 303: "See Other", 400: "Bad Request", 401: "Unauthorized", 413: "Payload Too Large"}
+        encoded = body.encode()
+        headers = {"Content-Length": str(len(encoded)), "Connection": "close"}
+        if extra:
+            headers.update(extra)
+        response = "HTTP/1.1 %d %s\r\n" % (status, reasons[status])
+        for key, value in headers.items():
+            response += "%s: %s\r\n" % (key, value)
+        try:
+            self._client.send(response.encode() + b"\r\n" + encoded)
+        except OSError:
+            pass
+        self._close()
+
+    def _close(self):
+        try:
+            self._client.close()
+        except OSError:
+            pass
+        self._client = None
+        self._buffer = b""
+
+
+def _parse_form(body):
+    result = {}
+    for pair in body.split("&"):
+        if "=" in pair:
+            key, value = pair.split("=", 1)
+            result[_unquote(key)] = _unquote(value)
+    return result
+
+
+def _unquote(value):
+    value = value.replace("+", " ")
+    parts = value.split("%")
+    result = parts[0]
+    for part in parts[1:]:
+        try:
+            result += chr(int(part[:2], 16)) + part[2:]
+        except ValueError:
+            result += "%" + part
+    return result
+
+
+def _checked(settings, name):
+    return " checked" if settings[name] else ""
+
+
+def _page(settings):
+    return """<!doctype html><meta name=viewport content='width=device-width,initial-scale=1'>
+<title>PolterHID</title><style>body{font:16px sans-serif;max-width:38rem;margin:2rem auto;padding:0 1rem}label{display:block;margin:.7rem 0}input[type=number]{width:6rem}button{padding:.6rem 1rem}</style>
+<h1>PolterHID</h1><p>Awareness-training HID configuration</p>
+<form method=post>
+<label><input type=checkbox name=enabled%s> Enable all injection</label>
+<label><input type=checkbox name=return_enabled%s> Inject Return key</label>
+<label>Return interval <input name=return_min_seconds type=number min=1 value=%d> to <input name=return_max_seconds type=number min=1 value=%d> seconds</label>
+<label><input type=checkbox name=mouse_enabled%s> Inject mouse jitter</label>
+<label>Mouse interval <input name=mouse_min_seconds type=number min=1 value=%d> to <input name=mouse_max_seconds type=number min=1 value=%d> seconds</label>
+<button>Save</button> <button name=action value=volume_up>Volume up</button> <button name=action value=volume_down>Volume down</button></form>
+<p><small>Changes take effect immediately. This HTTP interface has no TLS; use only a trusted training network.</small></p>""" % (
+        _checked(settings, "enabled"), _checked(settings, "return_enabled"),
+        settings["return_min_seconds"], settings["return_max_seconds"],
+        _checked(settings, "mouse_enabled"), settings["mouse_min_seconds"],
+        settings["mouse_max_seconds"],
+    )