audit.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 hashlib6import json7import os8import time9import uuid10from collections import deque11from pathlib import Path121314class Audit:15    def __init__(self, settings):16        self.settings = settings17        self.session_id = uuid.uuid4().hex18        self.path = Path(settings.log_dir) / self.session_id19        self.path.mkdir(parents=True, mode=0o700, exist_ok=True)20        os.chmod(self.path, 0o700)21        self.events = deque(maxlen=300)22        self._previous = "0" * 642324    def event(self, kind, **data):25        event = {"time": time.time(), "kind": kind, **data, "previous_hash": self._previous}26        raw = json.dumps(event, sort_keys=True, ensure_ascii=False, separators=(",", ":"))27        self._previous = hashlib.sha256(raw.encode()).hexdigest()28        event["hash"] = self._previous29        # Failure propagates: no HID execution without a working audit log.30        fd = os.open(self.path / "events.jsonl", os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)31        with os.fdopen(fd, "a", encoding="utf-8") as stream:32            stream.write(json.dumps(event, ensure_ascii=False) + "\n")33            stream.flush()34            os.fsync(stream.fileno())35        self.events.append(event)3637    def frame(self, frame):38        self.event("screenshot", sha256=frame.sha256, sequence=frame.sequence)39        if self.settings.log_screenshots:40            path = self.path / f"frame-{frame.sequence}-{frame.sha256[:12]}.png"41            try:42                fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)43            except FileExistsError:44                return45            with os.fdopen(fd, "wb") as output:46                output.write(frame.png)4748    def response(self, response):49        self.event("api_response", id=response.get("id"), status=response.get("status"),50                   usage=response.get("usage"), output_types=[x.get("type") for x in response.get("output", [])])51        if self.settings.log_api_responses:52            name = uuid.uuid4().hex + ".response.json"53            fd = os.open(self.path / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)54            with os.fdopen(fd, "w") as output:55                json.dump(response, output, ensure_ascii=False)

Projektquellen des Referenzstands. Konfiguration und Gerätezugriff erfolgen lokal auf dem Raspberry Pi.