QUELLCODE / PY

policy.py

src/hardware_box/safety/policy.py

1# OpenAss: nur privater, experimenteller Gebrauch; auf eigene Verantwortung.2# Niemals Unternehmensrechner, Produktivsysteme, Industrieanlagen oder3# sicherheitskritische Systeme anschließen oder bedienen.4# Haftungshinweis und gesetzliche Grenzen: ../../../DISCLAIMER.txt.5import asyncio6import time7from hardware_box.hid.keyboard import KeyboardLayout8910class SafetyError(RuntimeError):11    pass121314FIELDS = {15    "click": {"type", "x", "y", "button"}, "double_click": {"type", "x", "y", "button"},16    "move": {"type", "x", "y"}, "drag": {"type", "path"},17    "scroll": {"type", "x", "y", "scroll_x", "scroll_y"},18    "type": {"type", "text"}, "keypress": {"type", "keys"},19    "wait": {"type"}, "screenshot": {"type"},20}21MUTATIONS = set(FIELDS) - {"wait", "screenshot"}22ALIASES = {"CONTROL": "CTRL", "CTL": "CTRL", "COMMAND": "META", "CMD": "META",23           "GUI": "META", "WIN": "META", "WINDOWS": "META", "SUPER": "META",24           "OPTION": "ALT", "ESCAPE": "ESC", "DEL": "DELETE", "RETURN": "ENTER"}25for modifier in ("CTRL", "SHIFT", "ALT", "META"):26    for prefix in ("LEFT", "RIGHT", "L", "R"):27        ALIASES[prefix + modifier] = modifier28ALIASES.update({"ALTGR": "ALT", "LEFTCONTROL": "CTRL", "RIGHTCONTROL": "CTRL"})29DENIED = [{"CTRL", "ALT", "DELETE"}, {"META", "L"}, {"META", "R"},30          {"META", "X"}, {"CTRL", "ALT", "F1"}, {"CTRL", "ALT", "F2"},31          {"CTRL", "ALT", "F3"}, {"CTRL", "ALT", "F4"}, {"CTRL", "ALT", "F5"},32          {"CTRL", "ALT", "F6"}, {"CTRL", "SHIFT", "ESC"}, {"ALT", "F4"}]333435def integer(value, name, limit=None):36    if type(value) is not int or (limit is not None and abs(value) > limit):37        raise SafetyError(f"Ungültiger Integer: {name}")383940class Policy:41    def __init__(self, settings):42        self.settings = settings43        self.started = time.monotonic()44        self.actions = 045        self.last_action = 0.04647    def check_alive(self):48        if time.monotonic() - self.started >= self.settings.session_seconds:49            raise SafetyError("Sitzungslimit erreicht")5051    def validate(self, action, geometry):52        self.check_alive()53        if not isinstance(action, dict) or action.get("type") not in FIELDS:54            raise SafetyError("Unbekannte Aktion")55        kind = action["type"]56        if set(action) - FIELDS[kind]:57            raise SafetyError("Unerwartete Aktionsfelder")58        if kind in {"click", "double_click", "move", "scroll"}:59            self._point(action, geometry)60        if kind in {"click", "double_click"} and action.get("button", "left") not in {"left", "right", "middle"}:61            raise SafetyError("Maustaste nicht erlaubt")62        if kind == "scroll":63            integer(action.get("scroll_x", 0), "scroll_x", 1200)64            integer(action.get("scroll_y", 0), "scroll_y", 1200)65        if kind == "drag":66            path = action.get("path")67            if not isinstance(path, list) or not 2 <= len(path) <= 32:68                raise SafetyError("Drag benötigt 2 bis 32 Punkte")69            for point in path:70                if not isinstance(point, dict) or set(point) != {"x", "y"}:71                    raise SafetyError("Ungültiger Drag-Punkt")72                self._point(point, geometry)73        if kind == "type":74            value = action.get("text")75            if not isinstance(value, str) or not 1 <= len(value) <= self.settings.max_text_chars:76                raise SafetyError("Text fehlt oder überschreitet das Limit")77            layout = KeyboardLayout(self.settings.keyboard_layout)78            for stroke in layout.preflight_text(value):79                self._validate_stroke(stroke, layout)80        if kind == "keypress":81            keys = action.get("keys")82            if not isinstance(keys, list) or not 1 <= len(keys) <= 6 or any(not isinstance(k, str) for k in keys):83                raise SafetyError("Ungültige Tastenkombination")84            layout = KeyboardLayout(self.settings.keyboard_layout)85            stroke = layout.keypress(keys)86            self._validate_stroke(stroke, layout)87            if self.settings.allowed_keypresses:88                allowed = [layout.keypress(combo) for combo in self.settings.allowed_keypresses]89                if not any(self._modifiers(candidate.modifiers) == self._modifiers(stroke.modifiers) and set(candidate.usages) == set(stroke.usages) for candidate in allowed):90                    raise SafetyError("Tastenkombination nicht in der lokalen Freigabeliste")9192    @staticmethod93    def _modifiers(mask):94        return {name for name, bits in (("CTRL", 0x11), ("SHIFT", 0x22), ("ALT", 0x44), ("META", 0x88)) if mask & bits}9596    def _validate_stroke(self, stroke, layout):97        # Covers raw keypress aliases AND modifier/report combinations in typed text.98        for denied in [*DENIED, *self.settings.extra_denied_keypresses]:99            blocked = layout.keypress(list(denied))100            if self._modifiers(blocked.modifiers) <= self._modifiers(stroke.modifiers) and set(blocked.usages) <= set(stroke.usages):101                raise SafetyError("Tastenkombination verboten")102103    def _point(self, point, geometry):104        integer(point.get("x"), "x")105        integer(point.get("y"), "y")106        # Delegate authoritative black-bar and coordinate checks to Geometry.107        geometry.to_absolute(point["x"], point["y"])108109    async def reserve(self):110        self.check_alive()111        if self.actions >= self.settings.max_actions:112            raise SafetyError("Aktionslimit erreicht")113        delay = 1 / self.settings.max_actions_per_second - (time.monotonic() - self.last_action)114        if delay > 0:115            await asyncio.sleep(delay)116        self.check_alive()117        self.actions += 1118        self.last_action = time.monotonic()