commit d0c926abd6c10adf00d5b549e6d8cf5e134e83d3 Author: Commander1024 Date: Wed Aug 26 11:38:52 2026 +0200 Add PolterHID awareness training firmware diff --git a/README.md b/README.md new file mode 100644 index 0000000..eab22d3 --- /dev/null +++ b/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. diff --git a/hid.py b/hid.py new file mode 100644 index 0000000..7d9340b --- /dev/null +++ b/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))) diff --git a/main.py b/main.py new file mode 100644 index 0000000..27a166d --- /dev/null +++ b/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() diff --git a/settings.example.json b/settings.example.json new file mode 100644 index 0000000..4f5b956 --- /dev/null +++ b/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 +} diff --git a/settings.py b/settings.py new file mode 100644 index 0000000..40ae435 --- /dev/null +++ b/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) diff --git a/web.py b/web.py new file mode 100644 index 0000000..667d37c --- /dev/null +++ b/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 """ +PolterHID +

PolterHID

Awareness-training HID configuration

+
+ + + + + +
+

Changes take effect immediately. This HTTP interface has no TLS; use only a trusted training network.

""" % ( + _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"], + )