QUELLCODE / PY
app.py
src/hardware_box/web/app.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 json6import os7import secrets8from contextlib import asynccontextmanager9from pathlib import Path1011from fastapi import FastAPI, Request12from fastapi.responses import FileResponse, JSONResponse, Response13from pydantic import BaseModel, ConfigDict, Field, StrictBool1415from hardware_box.config import Settings16from hardware_box.capture.device import CaptureError17from hardware_box.hid.driver import HIDError18from hardware_box.controller.agent import ACTIVE19from hardware_box.openai.client import AstraClient, DemoAgent, APIError20from hardware_box.safety.policy import SafetyError212223class StartRequest(BaseModel):24 model_config = ConfigDict(extra="forbid")25 task: str = Field(min_length=1, max_length=4000)262728class Confirmation(BaseModel):29 model_config = ConfigDict(extra="forbid")30 nonce: str31 screenshot_sha256: str32 approved: StrictBool333435class ResultReview(BaseModel):36 model_config = ConfigDict(extra="forbid")37 verified: StrictBool383940def create_app(controller, token, manage_lifespan=True):41 if len(token) < 24:42 raise ValueError("BOX_WEB_TOKEN muss mindestens 24 Zeichen enthalten")43 settings = controller.settings44 asset_dir = Path(__file__).parent45 restart_required = False4647 @asynccontextmanager48 async def lifespan(app):49 if manage_lifespan:50 try:51 await controller.capture.start()52 except Exception as exc:53 controller.state, controller.error = "FAULT", str(exc)54 try:55 yield56 finally:57 if manage_lifespan:58 await controller.stop()59 await controller.hid.close()60 await controller.capture.close()6162 app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)6364 @app.middleware("http")65 async def security(request: Request, call_next):66 host = request.url.hostname67 if host not in {"127.0.0.1", "localhost", "::1"}:68 return JSONResponse({"detail": "Unzulässiger Host"}, status_code=403)69 origin = request.headers.get("origin")70 if origin and origin.rstrip("/") != str(request.base_url).rstrip("/"):71 return JSONResponse({"detail": "Fremder Origin blockiert"}, status_code=403)72 if request.url.path.startswith("/api/"):73 provided = request.headers.get("authorization", "")74 if not secrets.compare_digest(provided, "Bearer " + token):75 return JSONResponse({"detail": "Token fehlt oder ungültig"}, status_code=401)76 response = await call_next(request)77 response.headers["Cache-Control"] = "no-store"78 response.headers["X-Content-Type-Options"] = "nosniff"79 response.headers["Referrer-Policy"] = "no-referrer"80 response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' blob:; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'"81 return response8283 @app.exception_handler(SafetyError)84 @app.exception_handler(APIError)85 @app.exception_handler(HIDError)86 @app.exception_handler(CaptureError)87 @app.exception_handler(ValueError)88 async def safety_exception(request, exc):89 return JSONResponse({"detail": str(exc)}, status_code=409)9091 @app.get("/")92 async def index():93 return FileResponse(asset_dir / "index.html")9495 @app.get("/app.js")96 async def script():97 return FileResponse(asset_dir / "app.js", media_type="application/javascript")9899 @app.get("/style.css")100 async def style():101 return FileResponse(asset_dir / "style.css", media_type="text/css")102103 @app.get("/api/status")104 async def status():105 return {**controller.status(), "restart_required": restart_required}106107 @app.get("/api/screenshot")108 async def screenshot():109 # Pending screenshot is immutable and identified by its hash on approval.110 frame = controller.frame if controller.pending else await controller.capture.snapshot()111 return Response(frame.png, media_type="image/png", headers={"X-Screenshot-SHA256": frame.sha256})112113 @app.post("/api/start")114 async def start(request: StartRequest):115 if restart_required:116 raise SafetyError("Konfiguration gespeichert; Dienst zuerst neu starten")117 await controller.start(request.task)118 return controller.status()119120 @app.post("/api/pause")121 async def pause():122 await controller.stop(pause=True)123 return controller.status()124125 @app.post("/api/stop")126 async def stop():127 await controller.stop()128 return controller.status()129130 @app.post("/api/confirm")131 async def confirm(request: Confirmation):132 await controller.confirm(request.nonce, request.screenshot_sha256, request.approved)133 return {"accepted": True}134135 @app.post("/api/review")136 async def review(request: ResultReview):137 controller.review_result(request.verified)138 return controller.status()139140 @app.post("/api/manual")141 async def manual(request: dict):142 if restart_required:143 raise SafetyError("Neustart nach Konfigurationsänderung erforderlich")144 await controller.manual(request)145 return controller.status()146147 @app.post("/api/check")148 async def check():149 client = DemoAgent(settings) if settings.demo else AstraClient(settings, os.environ.get("OPENAI_API_KEY", ""))150 try:151 return await client.check()152 finally:153 await client.close()154155 @app.get("/api/logs")156 async def logs():157 return list(controller.audit.events) if controller.audit else []158159 @app.get("/api/settings")160 async def get_settings():161 return settings.model_dump()162163 @app.post("/api/settings")164 async def save_settings(changes: dict):165 async with controller._lifecycle:166 return await save_settings_locked(changes)167168 async def save_settings_locked(changes):169 nonlocal restart_required170 if controller.state in ACTIVE or (controller.worker and not controller.worker.done()):171 raise SafetyError("Vor Konfigurationsänderungen die Sitzung stoppen")172 allowed = {"capture_width", "capture_height", "screenshot_width", "screenshot_height",173 "target_width", "target_height", "keyboard_layout", "max_actions_per_second",174 "session_seconds", "max_actions", "max_text_chars", "extra_denied_keypresses", "allowed_keypresses"}175 if set(changes) - allowed:176 raise SafetyError("Diese Einstellung muss direkt in der lokalen Konfiguration geändert werden")177 values = settings.model_dump()178 values.update(changes)179 values["calibration_confirmed"] = False180 try:181 checked = Settings(**values)182 except ValueError as exc:183 raise SafetyError("Konfiguration enthält ungültige Werte") from exc184 path = Path(os.environ.get("BOX_CONFIG", "config/local.toml"))185 path.parent.mkdir(parents=True, exist_ok=True)186 lines = []187 for key, value in checked.model_dump().items():188 literal = str(value).lower() if isinstance(value, bool) else json.dumps(value)189 lines.append(f"{key} = {literal}")190 temporary = path.with_suffix(".tmp")191 fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)192 with os.fdopen(fd, "w") as stream:193 stream.write("\n".join(lines) + "\n")194 os.replace(temporary, path)195 restart_required = True196 return {"saved": True, "restart_required": True, "calibration_confirmed": False}197198 return app