Spaces:
Running
Running
| """Real Playwright adapter. Reuses accepted BrowserManager and P2 inventories. | |
| The default adapter is NOT a mock, does not start a browser and has no OS APIs. | |
| Canvas is disabled unless Center supplies exact reviewed fixture URLs on the server. | |
| """ | |
| from __future__ import annotations | |
| import base64 | |
| import binascii | |
| import hashlib | |
| import hmac | |
| import math | |
| import mimetypes | |
| import re | |
| import secrets | |
| import uuid | |
| from contextlib import asynccontextmanager | |
| from pathlib import Path | |
| from urllib.parse import unquote, urljoin, urlparse, urlsplit | |
| import httpx | |
| from web_guarded import GuardError | |
| INTERACTION_REVISION = 'semantic-context-r8-20260919' | |
| SEMANTIC_CLICK_ROLES = frozenset({'button', 'link', 'tab', 'menuitem', 'treeitem'}) | |
| SEMANTIC_CONTAINER_TAGS = frozenset({'div', 'span', 'li'}) | |
| CONSEQUENTIAL_LABEL = re.compile( | |
| r'\b(?:save|delete|remove|approve|sell|trade|logout|sign\s*out|' | |
| r'lưu|xóa|xoá|duyệt|mua|bán|gửi|xác\s*nhận|thanh\s*toán)\b', re.IGNORECASE) | |
| META_JS = r"""(el, contextState) => { | |
| const clean = (v,n=240) => String(v || '').replace(/\s+/g,' ').trim().slice(0,n); | |
| // Accessible labels are page text, not field values. In particular, never use | |
| // textContent on a referenced textarea/select/contenteditable control. | |
| const textWithoutValues = root => { | |
| if (!root || root.nodeType!==1 || root.isContentEditable || root.matches('input,textarea,select,output,script,style,[role="textbox"],[role="searchbox"],[contenteditable]:not([contenteditable="false"])')) return ''; | |
| const walker=root.ownerDocument.createTreeWalker(root,NodeFilter.SHOW_TEXT); | |
| let node, output='', visited=0; | |
| while ((node=walker.nextNode()) && visited++<512 && output.length<2000) { | |
| if (!node.parentElement?.isContentEditable && !node.parentElement?.closest('input,textarea,select,output,script,style,[role="textbox"],[role="searchbox"],[contenteditable]:not([contenteditable="false"])')) output+=' '+node.nodeValue; | |
| } | |
| return clean(output); | |
| }; | |
| // Snapshot-private WeakMap identities: no DOM attributes, globals or user values. | |
| const contextRoles = new Set(['region','group','dialog','tabpanel','tablist','toolbar', | |
| 'listitem','row','list','navigation','main','form','grid','tree','radiogroup']); | |
| const implicitRoles = {SECTION:'region',ARTICLE:'article',FIELDSET:'group',FIGURE:'figure', | |
| LI:'listitem',TR:'row',UL:'list',OL:'list',NAV:'navigation',MAIN:'main',FORM:'form',ASIDE:'complementary'}; | |
| const parentOf = node => node?.assignedSlot || node?.parentElement || node?.getRootNode?.()?.host || null; | |
| const contexts=[]; | |
| let contextAncestor=parentOf(el), contextDepth=0, contextsComplete=true; | |
| while(contextAncestor && contextDepth++<48) { | |
| const explicitRole=clean(contextAncestor.getAttribute('role'),80).toLowerCase().split(/\s+/)[0]; | |
| const cr=contextRoles.has(explicitRole) ? explicitRole : implicitRoles[contextAncestor.tagName]; | |
| if(cr) { | |
| if(contexts.length>=12) { contextsComplete=false; break; } | |
| const labelled=String(contextAncestor.getAttribute('aria-labelledby')||'').slice(0,1024).trim().split(/\s+/).filter(Boolean).slice(0,10); | |
| const contextRoot=contextAncestor.getRootNode(); | |
| const named=labelled.map(id=>textWithoutValues(contextRoot.getElementById?.(id)||contextAncestor.ownerDocument.getElementById(id))).join(' ').trim(); | |
| let contextLabel=clean(named || contextAncestor.getAttribute('aria-label')); | |
| let labelSource=named ? 'aria-labelledby' : contextLabel ? 'aria-label' : ''; | |
| if(!contextLabel) { | |
| // Read an explicit direct heading/legend/caption, never the entire card. | |
| const hs=Array.from(contextAncestor.children).filter(c=>/^(H[1-6]|LEGEND|FIGCAPTION|CAPTION)$/.test(c.tagName)).slice(0,2); | |
| if(hs.length===1) { contextLabel=textWithoutValues(hs[0]); labelSource='direct-heading'; } | |
| } | |
| if(!contextLabel && contextAncestor.id) { contextLabel=clean(contextAncestor.id); labelSource='dom-id'; } | |
| let identity=contextState.ids.get(contextAncestor); | |
| if(!identity) { identity=contextState.prefix+'_'+(++contextState.next); contextState.ids.set(contextAncestor,identity); } | |
| contexts.push({context_id:identity,tag:contextAncestor.tagName.toLowerCase(),role:cr, | |
| label:contextLabel,label_source:labelSource}); | |
| } | |
| contextAncestor=parentOf(contextAncestor); | |
| } | |
| if(contextAncestor) contextsComplete=false; | |
| const tag=el.tagName.toLowerCase(); | |
| const role=clean(el.getAttribute('role'),80).toLowerCase().split(/\s+/)[0]; | |
| const type=clean(el.type || el.getAttribute('type'),40).toLowerCase(); | |
| const labelIds=String(el.getAttribute('aria-labelledby') || '').slice(0,1024).trim().split(/\s+/).filter(Boolean).slice(0,10); | |
| const treeRoot=el.getRootNode(); | |
| const labelledBy=labelIds.map(id=>textWithoutValues(treeRoot.getElementById?.(id) || el.ownerDocument.getElementById(id))).join(' ').trim(); | |
| const labels=el.labels ? Array.from(el.labels).slice(0,10).map(textWithoutValues).join(' ') : ''; | |
| const label=clean(labelledBy || el.getAttribute('aria-label') || labels || el.getAttribute('title') || | |
| el.getAttribute('placeholder') || (tag==='img' ? el.getAttribute('alt') : '') || | |
| (['input','textarea','select'].includes(tag) ? '' : textWithoutValues(el)) || | |
| el.getAttribute('name') || el.id); | |
| const signature=clean([label,el.getAttribute('name'),el.id,el.getAttribute('autocomplete'),type].join(' '),1000); | |
| const box=el.getBoundingClientRect(); const style=getComputedStyle(el); | |
| let inheritedDisabled=false, ancestor=el, depth=0; | |
| while (ancestor && depth++<64) { | |
| if (ancestor.getAttribute?.('aria-disabled')==='true' || ancestor.hasAttribute?.('inert')) inheritedDisabled=true; | |
| ancestor=ancestor.parentElement || ancestor.getRootNode?.()?.host || null; | |
| } | |
| if (ancestor) inheritedDisabled=true; | |
| return {tag,role,type,label,signature,contexts,contexts_complete:contextsComplete,href:tag==='a' ? el.href : '', | |
| target:tag==='a' ? clean(el.getAttribute('target'),80) : '', | |
| expanded:el.getAttribute('aria-expanded'), selected:el.getAttribute('aria-selected'), | |
| haspopup:el.getAttribute('aria-haspopup'), pressed:el.getAttribute('aria-pressed'), | |
| download:tag==='a' ? el.hasAttribute('download') : false, | |
| accept:type==='file' ? clean(el.getAttribute('accept'),240) : '', | |
| multiple:type==='file' ? Boolean(el.multiple) : false, | |
| disabled:Boolean(el.disabled || el.matches(':disabled') || inheritedDisabled), | |
| readonly:Boolean(el.readOnly || el.getAttribute('aria-readonly')==='true'), connected:el.isConnected, | |
| visible:box.width>0 && box.height>0 && style.display!=='none' && style.visibility!=='hidden', | |
| scrollable_x:el.scrollWidth>el.clientWidth+1 && /^(auto|scroll|overlay)$/.test(style.overflowX), | |
| scrollable_y:el.scrollHeight>el.clientHeight+1 && /^(auto|scroll|overlay)$/.test(style.overflowY), | |
| checked:['checkbox','radio'].includes(type) ? Boolean(el.checked) : null, | |
| box:tag==='canvas' ? [box.x,box.y,box.width,box.height] : null}; | |
| }""" | |
| class BridgeRuntime: | |
| def __init__(self, manager, semantic, *, canvas_urls=()): | |
| from browser_core import ( | |
| MAX_UPLOAD_BYTES, _ALLOWED_DOWNLOAD_EXTENSIONS, _ALLOWED_UPLOAD_EXTENSIONS, | |
| _DANGEROUS_FIELD, _SENSITIVE_FIELD, _SUBMIT_WORDS, _content_type_allowed, | |
| _extension_allowed, _reject_executable_payload, download_chatgpt_attachment, | |
| ensure_public_web_url, safe_filename, | |
| ) | |
| self.manager, self.semantic = manager, semantic | |
| self.dangerous, self.sensitive, self.submit_words = _DANGEROUS_FIELD, _SENSITIVE_FIELD, _SUBMIT_WORDS | |
| self.ensure_url = ensure_public_web_url | |
| self.allowed_downloads = _ALLOWED_DOWNLOAD_EXTENSIONS | |
| self.allowed_uploads = _ALLOWED_UPLOAD_EXTENSIONS | |
| self.content_type_allowed = _content_type_allowed | |
| self.extension_allowed = _extension_allowed | |
| self.reject_executable = _reject_executable_payload | |
| self.download_chatgpt_attachment = download_chatgpt_attachment | |
| self.safe_filename = safe_filename | |
| self.max_upload_bytes = MAX_UPLOAD_BYTES | |
| self.max_inline_upload_bytes = 12 * 1024 | |
| self.max_download_bytes = 8 * 1024 * 1024 | |
| self.downloads = {} | |
| # This is trusted server configuration, NEVER a tool argument. | |
| self.canvas_urls = frozenset(canvas_urls) | |
| async def checkout(self, scope): | |
| async with self.manager.lease_scope(scope.lease_id): | |
| session = await self.manager.get(scope.owner, scope.session_id, touch=False) | |
| if getattr(session, 'lease', None) is None: | |
| raise GuardError('OWNED_SESSION_REQUIRED') | |
| async with session.lock: | |
| await self.revalidate(scope, session) | |
| yield session | |
| async def revalidate(self, scope, session): | |
| await self.manager.revalidate_locked_session(scope.owner, scope.session_id, session) | |
| async def dispose(self, snapshot): | |
| for handle in [snapshot.get('document'), snapshot.get('context_state'), *[r['handle'] for r in snapshot.get('refs',{}).values()]]: | |
| if handle is not None: | |
| try: | |
| await handle.dispose() | |
| except Exception: | |
| pass # Disposal of an already-destroyed JS context is benign. | |
| def action_timeout(action): | |
| return {'upload': 25, 'download': 20, 'navigate': 15, 'back': 12, 'click': 10, 'hover': 10}.get(action, 5) | |
| def _public_url(url): | |
| parts=urlsplit(str(url or '')) | |
| return f'{parts.scheme}://{parts.netloc}{parts.path}' | |
| async def release(self, scope): | |
| for key in [item for item in self.downloads if item[0] == scope.key]: | |
| self.downloads.pop(key,None) | |
| def _classify(self, meta): | |
| signature = meta['signature'] | |
| return {**meta, | |
| 'dangerous': meta['type'] == 'password' or bool(self.dangerous.search(signature)), | |
| 'sensitive': bool(self.sensitive.search(signature)), | |
| 'submit_like': meta['type'] in {'submit', 'reset'} or | |
| bool(CONSEQUENTIAL_LABEL.search(signature)) or | |
| (meta['tag'] != 'a' and bool(self.submit_words.search(signature)))} | |
| async def observe(self, session, page_id, frame_id, max_nodes, *, offset=0): | |
| if type(offset) is not int or not 0 <= offset <= 10000: | |
| raise GuardError('OBSERVATION_OFFSET_INVALID') | |
| pages, pmap, bp = self.semantic._page_inventory(session) | |
| if page_id: | |
| if page_id not in pmap: | |
| raise GuardError('PAGE_NOT_IN_OWNED_LEASE') | |
| else: | |
| ids = [i for i, p in pmap.items() if p is session.page] | |
| if len(ids) != 1: | |
| raise GuardError('EXPLICIT_PAGE_SELECTION_REQUIRED') | |
| page_id = ids[0] | |
| page = pmap[page_id] | |
| frames, fmap, bf = self.semantic._frame_inventory(session, page) | |
| if frame_id: | |
| if frame_id not in fmap: | |
| raise GuardError('FRAME_NOT_IN_OWNED_LEASE') | |
| frame = fmap[frame_id] | |
| else: | |
| frame = page.main_frame | |
| frame_id = self.semantic._stable_frame_id(session.session_id, frame) | |
| self.semantic._require_frame_origin(session, frame) | |
| document = await frame.evaluate_handle('document') | |
| selector = ('a[href],button,input,textarea,select,summary,img[alt],' | |
| '[role~="button"],[role~="link"],[role~="tab"],' | |
| '[role~="menuitem"],[role~="treeitem"],[role~="listbox"],' | |
| '[role~="grid"],[role~="tree"],[role~="region"],canvas') | |
| locator = frame.locator(selector) | |
| count = await locator.count() | |
| if offset > count: | |
| await document.dispose() | |
| raise GuardError('OBSERVATION_OFFSET_PAST_END') | |
| end = min(count, offset + max_nodes, 10000) | |
| context_state = await frame.evaluate_handle('(prefix)=>({ids:new WeakMap(),next:0,prefix})', 'ctx8_'+secrets.token_hex(12)) | |
| refs, controls = {}, [] | |
| private = {'page': page, 'frame': frame, 'page_id': page_id, 'frame_id': frame_id, | |
| 'document': document, 'context_state': context_state, 'refs': refs, 'page_map': pmap} | |
| try: | |
| for index in range(offset, end): | |
| handle = await locator.nth(index).element_handle(timeout=700) | |
| if handle is None: | |
| raise GuardError('CONTROL_REMOVED_DURING_OBSERVE') | |
| meta = self._classify(await handle.evaluate(META_JS, context_state)) | |
| if meta['dangerous']: | |
| controls.append({'ref_id': None, 'redacted': True, 'reason': 'AUTH_OR_HIGH_RISK'}) | |
| await handle.dispose() | |
| continue | |
| ref_id = 'ref_' + secrets.token_hex(16) | |
| refs[ref_id] = {'handle': handle, 'meta': meta} | |
| public = {k: meta[k] for k in ('tag','role','type','label','disabled','readonly','visible', | |
| 'checked','expanded','selected','haspopup','target','sensitive','submit_like', | |
| 'scrollable_x','scrollable_y','contexts','contexts_complete','pressed')} | |
| public['download_candidate'] = bool(meta['download'] and meta['href']) | |
| public['file_input'] = meta['type'] == 'file' | |
| public['accept'] = meta['accept'] if meta['type'] == 'file' else '' | |
| public['multiple'] = meta['multiple'] if meta['type'] == 'file' else False | |
| safe, _ = self.semantic._sanitize_value(public) | |
| controls.append({'ref_id': ref_id, **safe}) | |
| await self._document(session, private) | |
| except BaseException: | |
| await self.dispose(private) | |
| raise | |
| return private, {'page_id': page_id, 'frame_id': frame_id, 'page_inventory': pages, | |
| 'frame_inventory': frames, 'blocked_page_count': bp, 'blocked_frame_count': bf, | |
| 'controls': controls, 'truncated': offset > 0 or end < count, | |
| 'observed_count': len(controls), 'total_control_count': count, | |
| 'offset': offset, 'next_offset': end if end < min(count,10000) else None, | |
| 'window_mode': 'FRESH_DOM_SLICE_PREVIOUS_REFS_INVALIDATED', | |
| 'scan_limit': 10000, 'scan_limit_reached': count > 10000 and end == 10000, | |
| 'values_omitted': True, 'attachment_returned': False, | |
| 'interaction_revision': INTERACTION_REVISION, | |
| 'os_input_used': False} | |
| async def _document(self, session, snapshot): | |
| page, frame = snapshot['page'], snapshot['frame'] | |
| if page.is_closed() or page not in session.context.pages or frame not in page.frames: | |
| raise GuardError('PAGE_OR_FRAME_CLOSED') | |
| self.semantic._require_frame_origin(session, frame) | |
| try: | |
| same = await frame.evaluate('(oldDocument)=>oldDocument===document', snapshot['document']) | |
| except Exception as exc: | |
| raise GuardError('DOCUMENT_GENERATION_CHANGED') from exc | |
| if same is not True: | |
| raise GuardError('DOCUMENT_GENERATION_CHANGED') | |
| async def _target(self, session, snapshot, ref_id): | |
| await self._document(session, snapshot) | |
| row = snapshot['refs'].get(ref_id) | |
| if row is None: | |
| raise GuardError('REF_NOT_IN_SNAPSHOT') | |
| try: | |
| meta = self._classify(await row['handle'].evaluate(META_JS, snapshot['context_state'])) | |
| except Exception as exc: | |
| raise GuardError('REF_DETACHED_OR_CHANGED') from exc | |
| if not meta['connected']: | |
| raise GuardError('REF_DETACHED_OR_CHANGED') | |
| if meta['dangerous']: | |
| raise GuardError('AUTH_OR_HIGH_RISK_BLOCKED') | |
| if meta['contexts'] != row['meta']['contexts'] or meta['contexts_complete'] != row['meta']['contexts_complete']: | |
| raise GuardError('CONTROL_STATE_CHANGED_REOBSERVE') | |
| return row['handle'], meta | |
| async def validate(self, session, snapshot, ref_id, action, spec, *, sensitive_confirmed): | |
| if not isinstance(spec, dict): | |
| raise GuardError('SPEC_OBJECT_REQUIRED') | |
| await self._document(session, snapshot) | |
| if action == 'select_page': | |
| if set(spec) != {'page_id'} or ref_id: | |
| raise GuardError('SELECT_PAGE_SPEC_INVALID') | |
| _, pmap, _ = self.semantic._page_inventory(session) | |
| target = snapshot['page_map'].get(spec['page_id']) | |
| if target is None or pmap.get(spec['page_id']) is not target: | |
| raise GuardError('PAGE_IDENTITY_CHANGED') | |
| return | |
| if action == 'scroll': | |
| if (set(spec) not in ({'pixels'},{'pixels','axis'}) or type(spec.get('pixels')) is not int | |
| or not -5000 <= spec['pixels'] <= 5000 or spec.get('axis','y') not in {'x','y'}): | |
| raise GuardError('SCROLL_BOUND_INVALID') | |
| if not ref_id: | |
| return | |
| if action == 'navigate': | |
| if ref_id or set(spec) != {'url'} or not isinstance(spec['url'],str) or len(spec['url']) > 4096: | |
| raise GuardError('NAVIGATE_SPEC_INVALID') | |
| url = await self.ensure_url(spec['url'],require_https=True) | |
| self.manager.require_lease_target(session,url) | |
| return | |
| if action == 'back': | |
| if ref_id or spec: | |
| raise GuardError('BACK_SPEC_INVALID') | |
| return | |
| handle, meta = await self._target(session, snapshot, ref_id) | |
| original = snapshot['refs'][ref_id]['meta'] | |
| # Do not silently reacquire by label, CSS or coordinates after a DOM change. | |
| if meta != original: | |
| raise GuardError('CONTROL_STATE_CHANGED_REOBSERVE') | |
| if meta['disabled'] or not meta['visible']: | |
| raise GuardError('CONTROL_NOT_ACTIONABLE') | |
| if action == 'upload': | |
| route=set(spec) | |
| if route=={'files'}: | |
| files=spec['files'] | |
| if meta['type']!='file' or not isinstance(files,list) or len(files)!=1 or not isinstance(files[0],dict): | |
| raise GuardError('UPLOAD_SPEC_INVALID') | |
| item=files[0] | |
| if set(item)-{'download_url','file_id','mime_type','file_name'} or {'download_url','file_id'}-set(item): | |
| raise GuardError('UPLOAD_FILEPARAM_FIELDS_INVALID') | |
| if any(not isinstance(item.get(k,''),str) for k in ('download_url','file_id','mime_type','file_name')): | |
| raise GuardError('UPLOAD_FILEPARAM_TYPES_INVALID') | |
| if not item['download_url'].startswith('https://') or len(item['download_url'])>8192 or not item['file_id'].strip() or len(item['file_id'])>512: | |
| raise GuardError('UPLOAD_FILEPARAM_VALUES_INVALID') | |
| name=self.safe_filename(item.get('file_name',''),fallback='attachment.bin') | |
| if not self.extension_allowed(name,self.allowed_uploads): | |
| guessed=mimetypes.guess_extension(item.get('mime_type','').split(';',1)[0].strip()) or '' | |
| name=self.safe_filename('attachment'+guessed,fallback='attachment.bin') | |
| if not self.extension_allowed(name,self.allowed_uploads): | |
| raise GuardError('UPLOAD_FILE_TYPE_BLOCKED') | |
| elif route=={'inline_file'}: | |
| if meta['type']!='file': | |
| raise GuardError('UPLOAD_SPEC_INVALID') | |
| self._decode_inline_upload(spec['inline_file']) | |
| else: | |
| raise GuardError('UPLOAD_ROUTE_EXCLUSIVITY_REQUIRED') | |
| if sensitive_confirmed is not True: | |
| raise GuardError('EXACT_UPLOAD_CONFIRMATION_REQUIRED') | |
| return | |
| if action == 'download': | |
| if set(spec)!={'suggested_filename'} or not isinstance(spec['suggested_filename'],str) or len(spec['suggested_filename'])>255: | |
| raise GuardError('DOWNLOAD_SPEC_INVALID') | |
| if meta['tag']!='a' or not meta['download'] or not meta['href']: | |
| raise GuardError('OBSERVED_DOWNLOAD_ANCHOR_REQUIRED') | |
| url=await self.ensure_url(meta['href'],require_https=True) | |
| self.manager.require_lease_target(session,url) | |
| return | |
| if meta['submit_like'] or meta['type'] in {'submit','file','password'}: | |
| raise GuardError('CONSEQUENTIAL_ACTION_REQUIRES_EXISTING_REVIEWED_SUBMIT_ROUTE') | |
| if meta['sensitive'] and sensitive_confirmed is not True: | |
| raise GuardError('EXACT_SENSITIVE_DATA_CONFIRMATION_REQUIRED') | |
| if action == 'scroll': | |
| if (meta['tag'] not in {'div','section','ul','ol','main','aside'} or | |
| not meta.get('scrollable_'+spec.get('axis','y'))): | |
| raise GuardError('OBSERVED_SCROLLABLE_CONTAINER_REQUIRED') | |
| elif action in {'click', 'hover'}: | |
| native_target = meta['tag'] in {'a', 'button', 'summary'} | |
| semantic_target = (meta['tag'] in SEMANTIC_CONTAINER_TAGS and | |
| meta.get('role') in SEMANTIC_CLICK_ROLES) | |
| hover_image = action == 'hover' and meta['tag'] == 'img' and bool(meta['label']) | |
| # A forged role on an input/select/file/canvas never grants click authority. | |
| if spec or not (native_target or semantic_target or hover_image): | |
| raise GuardError('CLICK_NOT_SUPPORTED_USE_TYPED_ACTION') | |
| if meta['href']: | |
| url = await self.ensure_url(meta['href'], require_https=True) | |
| self.manager.require_lease_target(session, url) | |
| elif action == 'fill': | |
| if set(spec) != {'text'} or not isinstance(spec['text'],str) or len(spec['text']) > 8000: | |
| raise GuardError('FILL_SPEC_INVALID') | |
| if meta['tag'] not in {'input','textarea'} or meta['readonly']: | |
| raise GuardError('FIELD_NOT_EDITABLE') | |
| if meta['tag']=='input' and meta['type'] not in {'text','search','email','tel','url','number'}: | |
| raise GuardError('FIELD_TYPE_NOT_SUPPORTED') | |
| elif action == 'select': | |
| if set(spec) != {'value'} or not isinstance(spec['value'],str) or len(spec['value']) > 1000 or meta['tag'] != 'select': | |
| raise GuardError('SELECT_SPEC_INVALID') | |
| elif action == 'check': | |
| if set(spec) != {'checked'} or type(spec['checked']) is not bool or meta['type'] not in {'checkbox','radio'}: | |
| raise GuardError('CHECK_SPEC_INVALID') | |
| else: | |
| raise GuardError('ACTION_NOT_SUPPORTED') | |
| def _sha256(path): | |
| digest=hashlib.sha256() | |
| with Path(path).open('rb') as stream: | |
| for chunk in iter(lambda:stream.read(1024*1024),b''): | |
| digest.update(chunk) | |
| return digest.hexdigest() | |
| def _decode_inline_upload(self, item): | |
| if not isinstance(item,dict) or set(item)!={'file_name','content_base64','expected_sha256'}: | |
| raise GuardError('UPLOAD_INLINE_FIELDS_INVALID') | |
| if any(not isinstance(item[k],str) for k in ('file_name','content_base64','expected_sha256')): | |
| raise GuardError('UPLOAD_INLINE_TYPES_INVALID') | |
| if not item['file_name'] or len(item['file_name'])>255: | |
| raise GuardError('UPLOAD_INLINE_NAME_INVALID') | |
| name=self.safe_filename(item['file_name'],fallback='attachment.bin') | |
| if not self.extension_allowed(name,self.allowed_uploads): | |
| raise GuardError('UPLOAD_FILE_TYPE_BLOCKED') | |
| expected=item['expected_sha256'].casefold() | |
| if not re.fullmatch(r'[a-f0-9]{64}',expected): | |
| raise GuardError('UPLOAD_INLINE_SHA256_INVALID') | |
| encoded=item['content_base64'] | |
| max_encoded=((self.max_inline_upload_bytes+2)//3)*4 | |
| if not encoded or len(encoded)>max_encoded: | |
| raise GuardError('UPLOAD_INLINE_TOO_LARGE') | |
| try: | |
| raw=base64.b64decode(encoded.encode('ascii'),validate=True) | |
| except (UnicodeEncodeError,binascii.Error,ValueError) as exc: | |
| raise GuardError('UPLOAD_INLINE_BASE64_INVALID') from exc | |
| if not raw: | |
| raise GuardError('UPLOAD_INLINE_EMPTY') | |
| if len(raw)>self.max_inline_upload_bytes: | |
| raise GuardError('UPLOAD_INLINE_TOO_LARGE') | |
| actual=hashlib.sha256(raw).hexdigest() | |
| if not hmac.compare_digest(actual,expected): | |
| raise GuardError('UPLOAD_INLINE_SHA256_MISMATCH') | |
| if raw.startswith((b'MZ',b'\x7fELF')): | |
| raise GuardError('EXECUTABLE_PAYLOAD_BLOCKED') | |
| return name,raw,actual | |
| async def _download_public(self, session, scope, url, suggested_filename): | |
| current=await self.ensure_url(url,require_https=True) | |
| timeout=httpx.Timeout(20.0,connect=8.0,read=15.0,write=8.0,pool=8.0) | |
| headers={'Accept':'text/plain,text/csv,application/json,application/pdf,application/octet-stream;q=0.8,*/*;q=0.2', | |
| 'User-Agent':'NEXUS-Owned-Browser/1.0'} | |
| async with httpx.AsyncClient(timeout=timeout,follow_redirects=False,headers=headers,trust_env=False) as client: | |
| for hop in range(6): | |
| current=await self.ensure_url(current,require_https=True) | |
| self.manager.require_lease_target(session,current) | |
| async with client.stream('GET',current) as response: | |
| if response.status_code in {301,302,303,307,308}: | |
| location=str(response.headers.get('location') or '').strip() | |
| if not location or hop>=5: | |
| raise GuardError('DOWNLOAD_REDIRECT_BLOCKED') | |
| current=urljoin(current,location) | |
| continue | |
| if response.status_code!=200: | |
| raise GuardError('DOWNLOAD_HTTP_'+str(response.status_code)) | |
| disposition=str(response.headers.get('content-disposition') or '') | |
| match=re.search(r"filename\*?=(?:UTF-8''|\")?([^\";]+)",disposition,re.IGNORECASE) | |
| header_name=unquote(match.group(1).strip()) if match else '' | |
| url_name=Path(urlparse(current).path).name | |
| name=self.safe_filename(suggested_filename or header_name or url_name,fallback='download.bin') | |
| content_type=str(response.headers.get('content-type') or '').split(';',1)[0].strip().casefold() | |
| if not self.extension_allowed(name,self.allowed_downloads): | |
| guessed=mimetypes.guess_extension(content_type) or '' | |
| name=self.safe_filename('download'+guessed,fallback='download.bin') | |
| if not self.extension_allowed(name,self.allowed_downloads): | |
| raise GuardError('DOWNLOAD_FILE_TYPE_BLOCKED') | |
| if not self.content_type_allowed(content_type,name): | |
| raise GuardError('DOWNLOAD_CONTENT_TYPE_BLOCKED') | |
| declared=str(response.headers.get('content-length') or '').strip() | |
| if declared.isdigit() and int(declared)>self.max_download_bytes: | |
| raise GuardError('DOWNLOAD_TOO_LARGE') | |
| folder=session.folder/('download_guarded_'+uuid.uuid4().hex[:12]) | |
| folder.mkdir(parents=True,exist_ok=False) | |
| destination=folder/name | |
| size=0 | |
| try: | |
| with destination.open('xb') as output: | |
| async for chunk in response.aiter_bytes(): | |
| size+=len(chunk) | |
| if size>self.max_download_bytes: | |
| raise GuardError('DOWNLOAD_TOO_LARGE') | |
| output.write(chunk) | |
| if size<=0: | |
| raise GuardError('DOWNLOAD_EMPTY') | |
| self.reject_executable(destination) | |
| except BaseException: | |
| destination.unlink(missing_ok=True) | |
| try: folder.rmdir() | |
| except OSError: pass | |
| raise | |
| download_id='download_'+secrets.token_hex(16) | |
| file_sha=self._sha256(destination) | |
| self.downloads[(scope.key,download_id)]={'path':destination,'name':name,'size':size, | |
| 'sha256':file_sha,'content_type':content_type} | |
| return {'ok':True,'status':'PUBLIC_FILE_DOWNLOADED_ANONYMOUS','download_id':download_id, | |
| 'file_name':name,'file_size':size,'file_sha256':file_sha,'content_type':content_type, | |
| 'source_url':self._public_url(current),'redirect_count':hop, | |
| 'cookies_used':False,'authorization_header_used':False,'referer_header_used':False, | |
| 'public_anonymous_download':True,'write_may_have_applied':True} | |
| raise GuardError('DOWNLOAD_FAILED') | |
| def _origin_after(self, session, page, frame): | |
| if page.is_closed() or page not in session.context.pages or frame not in page.frames: | |
| raise GuardError('ACTION_CONTEXT_UNAVAILABLE_POST_DISPATCH') | |
| self.semantic._require_frame_origin(session, frame) | |
| async def perform(self, session, snapshot, ref_id, action, spec, *, scope): | |
| page,frame=snapshot['page'],snapshot['frame'] | |
| if action=='select_page': | |
| session.page=snapshot['page_map'][spec['page_id']] | |
| self._origin_after(session,session.page,session.page.main_frame) | |
| return {'ok':True,'status':'PAGE_SELECTED_NO_BRING_TO_FRONT','selected_page_id':spec['page_id'], 'write_may_have_applied':True} | |
| if action=='scroll': | |
| axis=spec.get('axis','y') | |
| delta={'left':spec['pixels'] if axis=='x' else 0, | |
| 'top':spec['pixels'] if axis=='y' else 0,'behavior':'instant'} | |
| if ref_id: | |
| target=snapshot['refs'][ref_id]['handle'] | |
| before=await target.evaluate('el=>({x:el.scrollLeft,y:el.scrollTop})') | |
| await target.evaluate('(el,delta)=>el.scrollBy(delta)',delta) | |
| after=await target.evaluate('el=>({x:el.scrollLeft,y:el.scrollTop})') | |
| else: | |
| before=await frame.evaluate('()=>({x:scrollX,y:scrollY})') | |
| await frame.evaluate('(delta)=>window.scrollBy(delta)',delta) | |
| after=await frame.evaluate('()=>({x:scrollX,y:scrollY})') | |
| self._origin_after(session,page,frame) | |
| return {'ok':True,'status':'SCROLL_READBACK','scroll_changed':before!=after,'write_may_have_applied':True} | |
| if action=='navigate': | |
| target=await self.ensure_url(spec['url'],require_https=True) | |
| self.manager.require_lease_target(session,target) | |
| await page.goto(target,wait_until='domcontentloaded',timeout=12000) | |
| current=await self.ensure_url(page.url,require_https=True) | |
| self.manager.require_lease_target(session,current) | |
| session.page=page | |
| self.semantic._require_frame_origin(session,page.main_frame) | |
| return {'ok':True,'status':'NAVIGATED_VERIFIED','current_url':self._public_url(current), | |
| 'verified':True,'write_may_have_applied':True} | |
| if action=='back': | |
| before=page.url | |
| await page.go_back(wait_until='domcontentloaded',timeout=10000) | |
| current=await self.ensure_url(page.url,require_https=True) | |
| self.manager.require_lease_target(session,current) | |
| session.page=page | |
| self.semantic._require_frame_origin(session,page.main_frame) | |
| return {'ok':True,'status':'BACK_NAVIGATION_VERIFIED','current_url':self._public_url(current), | |
| 'history_changed':current!=before,'verified':True,'write_may_have_applied':True} | |
| h=snapshot['refs'][ref_id]['handle'] | |
| if action=='upload': | |
| source_transport='OPENAI_FILEPARAMS_STREAMED_SERVER_SIDE' | |
| source_file_id_present=True | |
| raw_inline=None | |
| if set(spec)=={'files'}: | |
| item=spec['files'][0] | |
| name=self.safe_filename(item.get('file_name',''),fallback='attachment.bin') | |
| if not self.extension_allowed(name,self.allowed_uploads): | |
| guessed=mimetypes.guess_extension(item.get('mime_type','').split(';',1)[0].strip()) or '' | |
| name=self.safe_filename('attachment'+guessed,fallback='attachment.bin') | |
| elif set(spec)=={'inline_file'}: | |
| name,raw_inline,file_sha=self._decode_inline_upload(spec['inline_file']) | |
| source_transport='INLINE_BASE64_SHA256_BOUND' | |
| source_file_id_present=False | |
| else: | |
| raise GuardError('UPLOAD_ROUTE_EXCLUSIVITY_REQUIRED') | |
| folder=session.folder/('upload_guarded_'+uuid.uuid4().hex[:12]) | |
| folder.mkdir(parents=True,exist_ok=False) | |
| destination=folder/name | |
| try: | |
| if raw_inline is None: | |
| await self.download_chatgpt_attachment(download_url=item['download_url'],destination=destination, | |
| max_bytes=self.max_upload_bytes) | |
| else: | |
| destination.write_bytes(raw_inline) | |
| self.reject_executable(destination) | |
| size=destination.stat().st_size | |
| file_sha=self._sha256(destination) | |
| current_h,current_meta=await self._target(session,snapshot,ref_id) | |
| if current_h is not h or current_meta!=snapshot['refs'][ref_id]['meta']: | |
| raise GuardError('CONTROL_STATE_CHANGED_REOBSERVE') | |
| await h.set_input_files(str(destination),timeout=3000) | |
| observed=await h.evaluate("el=>Array.from(el.files||[],f=>({name:f.name,size:f.size}))") | |
| matched=observed==[{'name':name,'size':size}] | |
| except BaseException: | |
| destination.unlink(missing_ok=True) | |
| try: folder.rmdir() | |
| except OSError: pass | |
| raise | |
| self._origin_after(session,page,frame) | |
| return {'ok':matched,'status':'UPLOAD_ATTACHED_VERIFIED' if matched else 'UNKNOWN_POSTCONDITION', | |
| 'verified':matched,'file_name':name,'file_size':size,'file_sha256':file_sha, | |
| 'value_echoed':False,'source_transport':source_transport, | |
| 'source_file_id_present':source_file_id_present,'temporary_source_url_exposed':False, | |
| 'write_may_have_applied':True} | |
| if action=='download': | |
| return await self._download_public(session,scope,snapshot['refs'][ref_id]['meta']['href'], | |
| spec['suggested_filename']) | |
| if action=='hover': | |
| # This pointer belongs to the isolated browser, not the Windows desktop. | |
| await h.hover(timeout=5000) | |
| self._origin_after(session,page,frame) | |
| return {'ok':True,'status':'HOVER_APPLIED_REOBSERVE','verified':False, | |
| 'write_may_have_applied':True} | |
| if action=='click': | |
| before_url=page.url | |
| before_pages=len(session.context.pages) | |
| target = snapshot['refs'][ref_id]['meta'].get('target', '').casefold() | |
| if target == '_blank': | |
| # Subscribe before clicking: a new tab may arrive after click() returns. | |
| # Never select it or bring it to the foreground automatically. | |
| async with page.expect_popup(timeout=5000) as pending_popup: | |
| await h.click(timeout=5000) | |
| popup = await pending_popup.value | |
| await popup.wait_for_load_state('domcontentloaded', timeout=5000) | |
| else: | |
| await h.click(timeout=5000) | |
| if page.is_closed(): | |
| raise GuardError('SELECTED_PAGE_CLOSED_AFTER_CLICK') | |
| after_url=page.url | |
| if after_url!=before_url: | |
| current=await self.ensure_url(after_url,require_https=True) | |
| self.manager.require_lease_target(session,current) | |
| self.semantic._require_frame_origin(session,page.main_frame) | |
| else: | |
| self._origin_after(session,page,frame) | |
| return {'ok':True,'status':'CLICK_APPLIED_REOBSERVE','verified':False, | |
| 'navigation_observed':after_url!=before_url, | |
| 'new_page_opened':len(session.context.pages)>before_pages, | |
| 'write_may_have_applied':True} | |
| if action=='fill': | |
| await h.fill(spec['text'],timeout=2000) | |
| actual=await h.input_value(timeout=1000) | |
| expected=spec['text'] | |
| if snapshot['refs'][ref_id]['meta']['tag']=='textarea': | |
| actual=actual.replace('\r\n','\n').replace('\r','\n') | |
| expected=expected.replace('\r\n','\n').replace('\r','\n') | |
| matched=actual==expected | |
| elif action=='select': | |
| selected=await h.select_option(spec['value'],timeout=2000) | |
| actual=await h.evaluate('el=>Array.from(el.selectedOptions,o=>o.value)') | |
| matched=bool(selected) and actual==selected and actual==[spec['value']] | |
| elif action=='check': | |
| await h.set_checked(spec['checked'],timeout=2000) | |
| matched=await h.is_checked()==spec['checked'] | |
| else: | |
| raise GuardError('ACTION_NOT_SUPPORTED') | |
| self._origin_after(session,page,frame) | |
| return {'ok':matched,'status':'APPLIED_VERIFIED' if matched else 'UNKNOWN_POSTCONDITION', | |
| 'verified':matched,'value_echoed':False,'write_may_have_applied':True} | |
| async def read_download(self, session, scope, download_id, *, offset, max_bytes): | |
| if not re.fullmatch(r'download_[a-f0-9]{32}',str(download_id or '')): | |
| raise GuardError('DOWNLOAD_ID_INVALID') | |
| row=self.downloads.get((scope.key,download_id)) | |
| if row is None: | |
| raise GuardError('DOWNLOAD_NOT_FOUND_IN_OWNED_SCOPE') | |
| path=Path(row['path']) | |
| if path.is_symlink() or not path.is_file(): | |
| raise GuardError('DOWNLOAD_FILE_UNAVAILABLE') | |
| try: | |
| resolved=path.resolve(strict=True) | |
| root=Path(session.folder).resolve(strict=True) | |
| except OSError as exc: | |
| raise GuardError('DOWNLOAD_FILE_UNAVAILABLE') from exc | |
| if not resolved.is_relative_to(root): | |
| raise GuardError('DOWNLOAD_PATH_SCOPE_MISMATCH') | |
| size=resolved.stat().st_size | |
| if size!=row['size'] or self._sha256(resolved)!=row['sha256']: | |
| raise GuardError('DOWNLOAD_FILE_CHANGED') | |
| if offset>size: | |
| raise GuardError('DOWNLOAD_OFFSET_PAST_EOF') | |
| with resolved.open('rb') as stream: | |
| stream.seek(offset) | |
| chunk=stream.read(max_bytes) | |
| next_offset=offset+len(chunk) | |
| return {'ok':True,'status':'DOWNLOAD_CHUNK_READY','download_id':download_id, | |
| 'file_name':row['name'],'file_size':size,'file_sha256':row['sha256'], | |
| 'content_type':row['content_type'],'offset':offset,'next_offset':next_offset, | |
| 'eof':next_offset>=size,'chunk_base64':base64.b64encode(chunk).decode('ascii'), | |
| 'raw_chunk_bytes':len(chunk),'secret_values_exposed':False} | |
| async def matches(self, session, snapshot, ref_id, condition, expected): | |
| h,meta=await self._target(session,snapshot,ref_id) | |
| # Read-only matching of sensitive fields is not an extraction backdoor. | |
| if meta['sensitive']: | |
| raise GuardError('SENSITIVE_READBACK_NOT_EXPOSED') | |
| if condition=='checked' and type(expected) is bool and meta['type'] in {'checkbox','radio'}: | |
| return await h.is_checked()==expected | |
| if condition=='visible' and type(expected) is bool: | |
| return await h.is_visible()==expected | |
| if condition=='enabled' and type(expected) is bool: | |
| return (not meta['disabled'] and await h.is_enabled()) == expected | |
| if condition in {'expanded','aria_selected','aria_pressed'} and type(expected) is bool: | |
| attribute = {'expanded':'aria-expanded','aria_selected':'aria-selected','aria_pressed':'aria-pressed'}[condition] | |
| value = await h.get_attribute(attribute) | |
| return value in {'true','false'} and (value=='true') == expected | |
| if condition=='label_equals' and isinstance(expected,str) and len(expected)<=240: | |
| return meta['label'] == ' '.join(expected.split()) | |
| if condition=='input_equals' and isinstance(expected,str) and len(expected)<=8000 and meta['tag'] in {'input','textarea'}: | |
| return await h.input_value(timeout=700)==expected | |
| if condition=='selected_equals' and isinstance(expected,str) and meta['tag']=='select': | |
| return await h.evaluate('el=>Array.from(el.selectedOptions,o=>o.value)')==[expected] | |
| raise GuardError('WAIT_CONDITION_NOT_SUPPORTED') | |