QUELLCODE / PY
test_api_policy_web.py
tests/test_api_policy_web.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 time78import httpx9import pytest10from PIL import Image1112from hardware_box.config import Settings13from hardware_box.controller.agent import Controller14from hardware_box.capture.device import DemoCapture15from hardware_box.hid.driver import DryHID16from hardware_box.hid.mapping import Geometry17from hardware_box.openai.client import APIError, AstraClient, calls_from18from hardware_box.safety.policy import Policy, SafetyError19from hardware_box.vision.frame import prepare20from hardware_box.web.app import create_app212223@pytest.mark.parametrize("keys", [["CTRL", "ALT", "DELETE"], ["LEFT_CTRL", "RIGHT_ALT", "DEL"],24 ["right_gui", "L"], [" RCTRL ", " RIGHT_SHIFT ", "ESCAPE"],25 ["GUI", "R"], ["CTRL", "ALT", "SHIFT", "F1"]])26def test_denied_actual_report_aliases(keys):27 with pytest.raises(SafetyError):28 Policy(Settings()).validate({"type": "keypress", "keys": keys}, Geometry(1920, 1080, 3840, 2160))293031@pytest.mark.parametrize("action", [{"type": "click", "x": True, "y": 20},32 {"type": "click", "x": 1920, "y": 20},33 {"type": "type", "text": "x", "approved": True},34 {"type": "scroll", "x": 1, "y": 1, "scroll_y": 2000},35 {"type": "drag", "path": [{"x": 1, "y": 1}]}])36def test_invalid_or_extra_fields_rejected(action):37 with pytest.raises(ValueError if action.get("x") == 1920 else SafetyError):38 Policy(Settings()).validate(action, Geometry(1920, 1080, 3840, 2160))394041def test_letterbox_mapping_endpoints_and_bar_rejection():42 settings = Settings(screenshot_width=1920, screenshot_height=1080)43 frame = prepare(Image.new("RGB", (1024, 768)), settings, 1)44 assert frame.geometry.content_rect == (240, 0, 1440, 1080)45 assert frame.geometry.to_absolute(240, 0) == (0, 0)46 assert frame.geometry.to_absolute(1679, 1079) == (32767, 32767)47 with pytest.raises(ValueError):48 frame.geometry.to_absolute(239, 100)495051def test_extra_deny_applies_to_enter_embedded_in_text():52 policy = Policy(Settings(extra_denied_keypresses=[["ENTER"]]))53 with pytest.raises(SafetyError):54 policy.validate({"type": "type", "text": "hello\n"}, Geometry(1920,1080,1920,1080))555657def test_keypress_whitelist_keeps_builtin_denials():58 policy = Policy(Settings(allowed_keypresses=[["CTRL", "C"], ["CTRL", "ALT", "DEL"]]))59 geometry = Geometry(1920,1080,1920,1080)60 policy.validate({"type":"keypress","keys":["RIGHT_CTRL","C"]}, geometry)61 with pytest.raises(SafetyError):62 policy.validate({"type":"keypress","keys":["CTRL","V"]}, geometry)63 with pytest.raises(SafetyError):64 policy.validate({"type":"keypress","keys":["CTRL","ALT","DELETE"]}, geometry)656667@pytest.mark.parametrize("response", [{"status": "incomplete", "output": []},68 {"status": "completed", "output": [{"type": "function_call"}]},69 {"status": "completed", "output": [{"type": "computer_call", "call_id": "x", "action": {"type": "screenshot"}}]}])70def test_unknown_api_shapes_fail_closed(response):71 with pytest.raises(APIError):72 calls_from(response)737475async def test_stateless_api_replays_calls_reasoning_and_original_screenshots():76 requests = []77 def handler(request):78 body = json.loads(request.content)79 requests.append(body)80 if len(requests) == 1:81 return httpx.Response(200, json={"id": "r1", "status": "completed", "usage": {"total_tokens": 10}, "output": [82 {"type": "reasoning", "id": "reason1", "summary": [], "encrypted_content": "opaque"},83 {"type": "computer_call", "id": "c1", "call_id": "call1", "actions": [{"type": "screenshot"}], "pending_safety_checks": []}]})84 return httpx.Response(200, json={"id": "r2", "status": "completed", "usage": {"total_tokens": 20}, "output": [85 {"type": "message", "content": [{"type": "output_text", "text": "Finished"}]}]})86 settings = Settings()87 frame = prepare(Image.new("RGB", (1920,1080)), settings, 1)88 client = AstraClient(settings, "fake-unit-test-key", transport=httpx.MockTransport(handler))89 try:90 await client.start("Inspect", frame)91 await client.observe([{"type":"computer_call_output","call_id":"call1","output":{"type":"computer_screenshot","image_url":frame.data_url,"detail":"original"}}])92 assert requests[0]["tools"] == [{"type": "computer"}]93 assert requests[0]["model"] == "gpt-6-astra"94 assert requests[0]["store"] is False95 assert "previous_response_id" not in requests[1]96 assert requests[1]["input"][1]["encrypted_content"] == "opaque"97 assert requests[1]["input"][2]["call_id"] == requests[1]["input"][3]["call_id"]98 assert requests[1]["input"][3]["output"]["detail"] == "original"99 assert client.total_tokens == 30100 finally:101 await client.close()102103104async def test_api_error_not_retried():105 seen = []106 def handler(request):107 seen.append(request)108 return httpx.Response(429, json={"error": {"message": "rate limit"}})109 client = AstraClient(Settings(), "fake", transport=httpx.MockTransport(handler))110 try:111 with pytest.raises(APIError):112 await client._request()113 assert len(seen) == 1114 finally:115 await client.close()116117118async def test_web_token_origin_bounds_and_settings(tmp_path, monkeypatch):119 token = "test-token-" + "x" * 32120 settings = Settings(log_dir=str(tmp_path / "logs"))121 controller = Controller(settings, DemoCapture(settings), DryHID())122 monkeypatch.setenv("BOX_CONFIG", str(tmp_path / "local.toml"))123 app = create_app(controller, token, manage_lifespan=False)124 async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8080") as http:125 assert (await http.get("/")).status_code == 200126 assert (await http.get("/api/status")).status_code == 401127 headers = {"Authorization": "Bearer " + token}128 assert (await http.get("/api/status", headers=headers)).status_code == 200129 assert (await http.post("/api/stop", headers={**headers,"Origin":"https://evil.example"}, json={})).status_code == 403130 assert (await http.get("/api/status", headers={**headers,"Host":"evil.example"})).status_code == 403131 screenshot = await http.get("/api/screenshot", headers=headers)132 assert screenshot.headers["Content-Security-Policy"].find("frame-ancestors 'none'") >= 0133 assert screenshot.content[:8] == b"\x89PNG\r\n\x1a\n"134 assert (await http.post("/api/manual", headers=headers, json={"type":"keypress","keys":["LEFT_CTRL","LEFT_ALT","DEL"]})).status_code == 409135 assert (await http.post("/api/settings", headers=headers, json={"demo":False})).status_code == 409136 assert (await http.post("/api/settings", headers=headers, json={"screenshot_width":1280})).status_code == 200137 assert (tmp_path / "local.toml").exists()138 assert (await http.post("/api/start", headers=headers, json={"task":"Inspect"})).status_code == 409