File size: 7,103 Bytes
795f737 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | """Validated executor. Permissions and completion evidence come from trusted code."""
from collections import Counter
from dataclasses import dataclass
from time import perf_counter, process_time
from urllib.parse import urlsplit
from .actions import Kind, TARGETED
from .authority import Rejected
from .state import digest
@dataclass(frozen=True)
class Outcome:
status: str
code: str
wall_ms: float
cpu_ms: float
evidence: str | None = None
class Runtime:
def __init__(self, browser, authority, permission, completion=None, recorder=None):
self.browser = browser
self.authority = authority
# Callbacks are configured by the host, never parsed from page/model text.
self.permission = permission
self.completion = completion
self.completion_task = authority.task_id
self.recorder = recorder
self.failures = Counter()
self.attempts = Counter()
self.max_steps = 100
def execute(self, decision):
started, cpu = perf_counter(), process_time()
evidence = None
status, code = "rejected", "UNINITIALIZED"
action = decision.action
target = None
state = self.browser.state
with self.authority.lock:
try:
if state is None:
raise Rejected("OBSERVATION_REQUIRED")
self.authority.validate(decision.ticket, state)
self.browser.validate_document()
key = (self.authority.task_id, state.state_hash, self.action_key(action, state))
if self.authority.sequence >= self.max_steps:
raise Rejected("STEP_BUDGET")
if self.failures[key] >= 2 or self.attempts[key] >= 3:
raise Rejected("REPEATED_ACTION")
if not self.permission(self.authority.goal, state, action):
raise Rejected("HOST_POLICY_DENIED")
if action.kind in {Kind.NAVIGATE, Kind.OPEN_TAB}:
parsed = urlsplit(action.args[0])
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
raise Rejected("UNSAFE_URL")
if action.kind in TARGETED:
target, element = self.browser.resolve(action.args[0])
if element.sensitive:
raise Rejected("SENSITIVE_TARGET_REQUIRES_HOST_FLOW")
if action.kind == Kind.TYPE and element.role not in {"textbox", "searchbox", "combobox", "spinbutton"}:
raise Rejected("TYPE_ROLE_MISMATCH")
if action.kind == Kind.SELECT and element.role not in {"combobox", "listbox"}:
raise Rejected("SELECT_ROLE_MISMATCH")
if action.kind == Kind.FINISH:
if not self.completion or self.completion_task != self.authority.task_id:
raise Rejected("NO_COMPLETION_VERIFIER")
evidence = self.completion(self.browser)
if not isinstance(evidence, str) or not evidence:
raise Rejected("COMPLETION_NOT_PROVEN")
self.authority.consume() # Attempt consumes ticket even if browser action fails.
self.attempts[key] += 1
try:
self._dispatch(action, target)
status, code = "ok", action.kind.name
except Rejected:
self.failures[key] += 1
raise
except Exception as exc:
self.failures[key] += 1
# Exceptions often contain typed values, URLs or page text. Do not log them.
status, code = "failed", type(exc).__name__
except Rejected as exc:
status, code = "rejected", str(exc)
finally:
if target:
target.dispose()
outcome = Outcome(status, code, (perf_counter()-started)*1000, (process_time()-cpu)*1000, evidence)
if self.recorder:
self.recorder.record(decision, state, outcome)
return outcome
@staticmethod
def action_key(action, state):
args = list(action.args)
if action.kind in TARGETED:
element = next((e for e in state.elements if e.ref == args[0]), None)
args[0] = element.signature if element else args[0]
return digest([action.kind, args])
def _dispatch(self, action, target):
k, args, page = action.kind, action.args, self.browser.page
if k == Kind.CLICK:
target.click(timeout=2000)
elif k == Kind.TYPE:
target.fill(args[1], timeout=2000)
elif k == Kind.SELECT:
target.select_option(value=args[1], timeout=2000)
elif k == Kind.SCROLL:
page.mouse.wheel(*args)
elif k == Kind.WAIT:
page.wait_for_timeout(args[0])
elif k == Kind.NAVIGATE:
page.goto(args[0], wait_until="domcontentloaded", timeout=10000)
elif k == Kind.BACK:
page.go_back(wait_until="domcontentloaded", timeout=10000)
elif k == Kind.FORWARD:
page.go_forward(wait_until="domcontentloaded", timeout=10000)
elif k == Kind.PRESS_KEY:
if args[1] not in {"Enter", "Tab", "Escape", "ArrowUp", "ArrowDown", "ArrowLeft", "ArrowRight", "Space", "Home", "End"}:
raise Rejected("UNSUPPORTED_KEY")
target.press(args[1], timeout=2000)
elif k == Kind.SUBMIT:
target.press("Enter", timeout=2000)
elif k == Kind.OPEN_TAB:
self.browser.page = self.browser.context.new_page()
self.browser.page.goto(args[0], wait_until="domcontentloaded", timeout=10000)
self.browser.sync_tabs()
self.browser.state = None
elif k in {Kind.CLOSE_TAB, Kind.FOCUS_TAB}:
self.browser.sync_tabs()
if args[0] not in self.browser.tabs:
raise Rejected("UNKNOWN_TAB")
requested = self.browser.tabs[args[0]]
if k == Kind.CLOSE_TAB:
if len(self.browser.tabs) <= 1:
raise Rejected("LAST_TAB")
requested.close()
self.browser.sync_tabs()
if requested == self.browser.page:
self.browser.page = next(iter(self.browser.tabs.values()))
else:
self.browser.page = requested
requested.bring_to_front()
self.browser.state = None
elif k == Kind.EXTRACT:
# Result remains in task memory, not persistent telemetry by default.
self.extracted = target.inner_text(timeout=2000)[:16384]
elif k == Kind.RECOVER:
self.browser.observe()
elif k == Kind.ASK_USER:
self.question = args[0]
elif k == Kind.FINISH:
pass
else:
raise Rejected("UNSUPPORTED_ACTION")
|