PolterHID/web.py

"""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"],
    )