from __future__ import annotations import asyncio import hashlib import hmac import json import re import time import uuid from dataclasses import dataclass from typing import Any, Callable from urllib.parse import urlparse from browser_core import ( BrowserBridgeError, canonical_https_origin, inventory_location, redact_sensitive_query_values, ) VERSION = "0.2.0-field-safe-blocks-r5" CURSOR_TTL_SECONDS = 60.0 MAX_CURSOR_RECORDS = 256 MIN_OUTPUT_CHARS = 4_000 MAX_OUTPUT_CHARS = 60_000 DEFAULT_OUTPUT_CHARS = 36_000 MAX_TEXT_CHUNK = 2_000 MAX_CELL_CHARS = 512 MAX_RECORD_CHARS = 7_000 MAX_TABLES = 25 MAX_ROWS_PER_TABLE = 1_000 MAX_COLUMNS = 60 MAX_LISTS = 40 MAX_LIST_ITEMS = 500 MAX_LINKS = 750 MAX_JSON_LD = 40 _CURSOR_RE = re.compile(r"^cur_[a-f0-9]{32}$") _PAGE_RE = re.compile(r"^pg_[a-f0-9]{24}$") _FRAME_RE = re.compile(r"^fr_[a-f0-9]{24}$") _AUTH_LINE_RE = re.compile( r"password|mật\s*khẩu|passcode|otp|one.?time|verification.?code|auth.?code|" r"access.?token|refresh.?token|secret|credential", re.IGNORECASE, ) _NEXT_RE = re.compile(r"^(next|next page|older|more|tiếp|trang sau|sau)$", re.IGNORECASE) @dataclass class CursorRecord: cursor_id: str owner: str session_id: str lease_digest: str snapshot_id: str page_id: str frame_id: str mode: str max_output_chars: int records: tuple[dict[str, Any], ...] offset: int metadata: dict[str, Any] source_fingerprint: str created: float expires: float state: str = "ACTIVE" error_code: str = "" class SemanticCursorStore: """Server-held one-use cursors over one frozen semantic snapshot.""" def __init__( self, *, ttl_seconds: float = CURSOR_TTL_SECONDS, max_records: int = MAX_CURSOR_RECORDS, clock: Callable[[], float] = time.monotonic, ) -> None: self.ttl_seconds = float(ttl_seconds) self.max_records = int(max_records) self.clock = clock self.records: dict[str, CursorRecord] = {} self.lock = asyncio.Lock() def _purge(self, now: float) -> None: for key, row in list(self.records.items()): if now >= row.expires + 300: self.records.pop(key, None) if len(self.records) < self.max_records: return retired = sorted( (r for r in self.records.values() if r.state != "ACTIVE" or now >= r.expires), key=lambda r: (r.expires, r.created, r.cursor_id), ) for row in retired: self.records.pop(row.cursor_id, None) if len(self.records) < self.max_records: return @staticmethod def _lease_digest(lease_id: str) -> str: return hashlib.sha256(str(lease_id or "").encode("utf-8")).hexdigest() @staticmethod def _measure(value: Any) -> int: return len(json.dumps(value, ensure_ascii=False, separators=(",", ":"))) @classmethod def _slice(cls, records: tuple[dict[str, Any], ...], start: int, max_chars: int) -> tuple[list[dict[str, Any]], int]: selected: list[dict[str, Any]] = [] used = 2 index = start while index < len(records): item = records[index] size = cls._measure(item) + (1 if selected else 0) if selected and used + size > max_chars: break if not selected and size > max_chars: # Records are normalized below to stay under MAX_RECORD_CHARS. This is a final fail-closed guard. raise BrowserBridgeError( "READ_DATA_RECORD_TOO_LARGE", "One normalized semantic record exceeds the requested output bound.", ) selected.append(item) used += size index += 1 return selected, index async def start( self, *, owner: str, session_id: str, lease_id: str, page_id: str, frame_id: str, mode: str, max_output_chars: int, records: list[dict[str, Any]], metadata: dict[str, Any], source_fingerprint: str, ) -> dict[str, Any]: frozen = tuple(records) items, next_offset = self._slice(frozen, 0, max_output_chars) snapshot_id = "snap_" + uuid.uuid4().hex next_cursor = None now = self.clock() if next_offset < len(frozen): async with self.lock: self._purge(now) if len(self.records) >= self.max_records: raise BrowserBridgeError("READ_DATA_CURSOR_CAPACITY", "No semantic-read cursor capacity is available.") next_cursor = "cur_" + uuid.uuid4().hex self.records[next_cursor] = CursorRecord( cursor_id=next_cursor, owner=owner.casefold(), session_id=session_id, lease_digest=self._lease_digest(lease_id), snapshot_id=snapshot_id, page_id=page_id, frame_id=frame_id, mode=mode, max_output_chars=max_output_chars, records=frozen, offset=next_offset, metadata=dict(metadata), source_fingerprint=source_fingerprint, created=now, expires=now + self.ttl_seconds, ) return { "snapshot_id": snapshot_id, "items": items, "returned_count": len(items), "total_record_count": len(frozen), "page_index": 1, "next_cursor": next_cursor, "cursor_expires_in_ms": round(self.ttl_seconds * 1000) if next_cursor else None, "source_fingerprint": source_fingerprint, "metadata": metadata, } async def continue_page( self, *, cursor_id: str, owner: str, session_id: str, lease_id: str, max_output_chars: int, ) -> dict[str, Any]: if not _CURSOR_RE.fullmatch(str(cursor_id or "")): raise BrowserBridgeError("READ_DATA_CURSOR_INVALID", "The semantic-read cursor is invalid.") now = self.clock() async with self.lock: self._purge(now) row = self.records.get(cursor_id) if row is None: raise BrowserBridgeError("READ_DATA_CURSOR_NOT_FOUND", "The semantic-read cursor is unavailable.") if row.state != "ACTIVE": raise BrowserBridgeError(row.error_code or "READ_DATA_CURSOR_ALREADY_USED", "The semantic-read cursor cannot be replayed.") if now >= row.expires: row.state = "EXPIRED" row.error_code = "READ_DATA_CURSOR_EXPIRED" raise BrowserBridgeError(row.error_code, "The semantic-read cursor expired.") if not hmac.compare_digest(row.owner, owner.casefold()): raise BrowserBridgeError("READ_DATA_CURSOR_OWNER_MISMATCH", "The semantic-read cursor owner does not match.") if not hmac.compare_digest(row.session_id, session_id): raise BrowserBridgeError("READ_DATA_CURSOR_SESSION_MISMATCH", "The semantic-read cursor session does not match.") if not hmac.compare_digest(row.lease_digest, self._lease_digest(lease_id)): raise BrowserBridgeError("READ_DATA_CURSOR_LEASE_MISMATCH", "The semantic-read cursor lease does not match.") if row.max_output_chars != max_output_chars: raise BrowserBridgeError("READ_DATA_CURSOR_BOUND_MISMATCH", "Keep max_output_chars unchanged while continuing a cursor.") items, next_offset = self._slice(row.records, row.offset, max_output_chars) row.state = "USED" row.error_code = "READ_DATA_CURSOR_ALREADY_USED" next_cursor = None if next_offset < len(row.records): if len(self.records) >= self.max_records: self._purge(now) if len(self.records) >= self.max_records: raise BrowserBridgeError("READ_DATA_CURSOR_CAPACITY", "No semantic-read cursor capacity is available.") next_cursor = "cur_" + uuid.uuid4().hex self.records[next_cursor] = CursorRecord( cursor_id=next_cursor, owner=row.owner, session_id=row.session_id, lease_digest=row.lease_digest, snapshot_id=row.snapshot_id, page_id=row.page_id, frame_id=row.frame_id, mode=row.mode, max_output_chars=row.max_output_chars, records=row.records, offset=next_offset, metadata=dict(row.metadata), source_fingerprint=row.source_fingerprint, created=row.created, expires=row.expires, ) return { "snapshot_id": row.snapshot_id, "items": items, "returned_count": len(items), "total_record_count": len(row.records), "page_index": (row.offset // max(1, len(items))) + 1 if items else 1, "next_cursor": next_cursor, "cursor_expires_in_ms": max(0, round((row.expires - now) * 1000)) if next_cursor else None, "source_fingerprint": row.source_fingerprint, "metadata": dict(row.metadata), } class SemanticReadData: """Lease-bound, read-only semantic extraction for one explicit page/frame. This does not scroll, click pagination controls, change the selected page, take screenshots, or return attachments. It reports current-DOM completeness separately from logical-dataset completeness so virtualized/paginated data is never silently promoted to complete. """ def __init__(self, manager: Any, *, clock: Callable[[], float] = time.monotonic) -> None: self.manager = manager self.clock = clock self.cursors = SemanticCursorStore(clock=clock) self._page_ids: dict[str, dict[int, str]] = {} self._frame_ids: dict[str, dict[int, str]] = {} @staticmethod def _new_id(prefix: str) -> str: return prefix + hashlib.sha256(uuid.uuid4().bytes).hexdigest()[:24] def _stable_page_id(self, session_id: str, page: Any) -> str: mapping = self._page_ids.setdefault(session_id, {}) key = id(page) if key not in mapping: mapping[key] = self._new_id("pg_") return mapping[key] def _stable_frame_id(self, session_id: str, frame: Any) -> str: mapping = self._frame_ids.setdefault(session_id, {}) key = id(frame) if key not in mapping: mapping[key] = self._new_id("fr_") return mapping[key] def _page_inventory(self, session: Any) -> tuple[list[dict[str, Any]], dict[str, Any], int]: live = [] by_id = {} blocked_count = 0 allowed = set(getattr(getattr(session, "lease", None), "allowed_origins", ()) or ()) for page in list(session.context.pages): try: if page.is_closed(): continue origin, path = inventory_location(page.url) if origin not in allowed: blocked_count += 1 continue page_id = self._stable_page_id(session.session_id, page) row = { "page_id": page_id, "current_origin": origin, "current_path": path, "query_fragment_title_text_and_fields_omitted": True, } live.append(row) by_id[page_id] = page except Exception: blocked_count += 1 continue live.sort(key=lambda row: row["page_id"]) return live, by_id, blocked_count def _frame_inventory(self, session: Any, page: Any) -> tuple[list[dict[str, Any]], dict[str, Any], int]: rows = [] by_id = {} blocked_count = 0 allowed = set(getattr(getattr(session, "lease", None), "allowed_origins", ()) or ()) frames = list(page.frames) for frame in frames: try: effective_url = str(frame.url or "") if effective_url == "about:blank" and frame.parent_frame is not None: effective_url = str(frame.parent_frame.url or "") origin, path = inventory_location(effective_url) if origin not in allowed: blocked_count += 1 continue frame_id = self._stable_frame_id(session.session_id, frame) parent = frame.parent_frame parent_id = None if parent is not None: parent_url = str(parent.url or "") if parent_url == "about:blank" and parent.parent_frame is not None: parent_url = str(parent.parent_frame.url or "") parent_origin, _parent_path = inventory_location(parent_url) if parent_origin in allowed: parent_id = self._stable_frame_id(session.session_id, parent) row = { "frame_id": frame_id, "parent_frame_id": parent_id, "main_frame": frame is page.main_frame, "current_origin": origin, "current_path": path, "query_fragment_text_and_fields_omitted": True, } rows.append(row) by_id[frame_id] = frame except Exception: blocked_count += 1 continue rows.sort(key=lambda row: (not row["main_frame"], row["frame_id"])) return rows, by_id, blocked_count @staticmethod def _require_frame_origin(session: Any, frame: Any) -> None: lease = getattr(session, "lease", None) if lease is None: raise BrowserBridgeError("READ_DATA_LEASE_REQUIRED", "Semantic read_data is available only on an owned leased session.") raw = str(frame.url or "") if raw == "about:blank" and frame.parent_frame is not None: raw = str(frame.parent_frame.url or "") try: origin = canonical_https_origin(raw, origin_only=False) except Exception as exc: raise BrowserBridgeError("READ_DATA_FRAME_ORIGIN_UNAVAILABLE", "The selected frame origin cannot be verified.") from exc if origin not in lease.allowed_origins: raise BrowserBridgeError("READ_DATA_FRAME_ORIGIN_BLOCKED", "The selected frame is outside the task lease origin set.") @staticmethod def _sanitize_string(value: str) -> tuple[str, int]: text = str(value or "") omitted = 0 lines = [] for line in text.splitlines() or [text]: if _AUTH_LINE_RE.search(line): lines.append("[AUTH_FIELD_OMITTED]") omitted += 1 else: lines.append(line) return "\n".join(lines), omitted @classmethod def _sanitize_value(cls, value: Any) -> tuple[Any, int]: if isinstance(value, str): return cls._sanitize_string(value) if isinstance(value, list): out = [] total = 0 for item in value: cleaned, count = cls._sanitize_value(item) out.append(cleaned) total += count return out, total if isinstance(value, dict): out: dict[str, Any] = {} total = 0 for key, item in value.items(): if _AUTH_LINE_RE.search(str(key)): out[str(key)] = "[AUTH_FIELD_OMITTED]" total += 1 continue cleaned, count = cls._sanitize_value(item) out[str(key)] = cleaned total += count return out, total return value, 0 @staticmethod def _bounded_record(record: dict[str, Any]) -> dict[str, Any]: candidate = dict(record) raw = json.dumps(candidate, ensure_ascii=False, separators=(",", ":")) if len(raw) <= MAX_RECORD_CHARS: return candidate if candidate.get("kind") == "table_row": cells = list(candidate.get("cells") or []) while cells and len(json.dumps({**candidate, "cells": cells}, ensure_ascii=False, separators=(",", ":"))) > MAX_RECORD_CHARS: cells.pop() candidate["cells"] = cells candidate["record_truncated"] = True candidate["incomplete_reason"] = "row_output_bound" return candidate candidate = { "kind": str(candidate.get("kind") or "record"), "record_truncated": True, "incomplete_reason": "record_output_bound", "sha256": hashlib.sha256(raw.encode("utf-8")).hexdigest(), } return candidate @staticmethod def _records(extracted: dict[str, Any]) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] text = str(extracted.get("page_text") or "") for index in range(0, len(text), MAX_TEXT_CHUNK): records.append({"kind": "page_text", "chunk_index": (index // MAX_TEXT_CHUNK) + 1, "text": text[index:index + MAX_TEXT_CHUNK]}) for table in list(extracted.get("tables") or []): table_index = int(table.get("table_index") or 0) headers = [str(x)[:MAX_CELL_CHARS] for x in list(table.get("headers") or [])[:MAX_COLUMNS]] records.append({ "kind": "table_header", "table_index": table_index, "table_kind": str(table.get("table_kind") or ""), "caption": str(table.get("caption") or "")[:1000], "headers": headers, "observed_row_count": int(table.get("row_count") or 0), "declared_row_count": table.get("declared_row_count"), "column_count": int(table.get("column_count") or len(headers)), }) for row_index, row in enumerate(list(table.get("rows") or []), start=1): cells = [str(x)[:MAX_CELL_CHARS] for x in list(row)[:MAX_COLUMNS]] if any(_AUTH_LINE_RE.search(cell) for cell in cells): records.append({ "kind": "auth_omitted", "source_kind": "table_row", "table_index": table_index, "row_index": row_index, "value": "[AUTH_FIELD_OMITTED]", }) else: records.append({"kind": "table_row", "table_index": table_index, "row_index": row_index, "cells": cells}) for row in list(extracted.get("lists") or []): list_index = int(row.get("list_index") or 0) for item_index, item in enumerate(list(row.get("items") or []), start=1): records.append({"kind": "list_item", "list_index": list_index, "item_index": item_index, "text": str(item)[:MAX_TEXT_CHUNK]}) for index, link in enumerate(list(extracted.get("links") or []), start=1): records.append({ "kind": "link", "index": index, "text": str(link.get("text") or "")[:1000], "url": redact_sensitive_query_values(str(link.get("url") or ""))[:3000], }) for index, value in enumerate(list(extracted.get("json_ld") or []), start=1): records.append({"kind": "json_ld", "index": index, "value": value}) return [SemanticReadData._bounded_record(item) for item in records] async def _extract_frame(self, frame: Any, mode: str) -> dict[str, Any]: script = r""" ({ mode, maxTables, maxRows, maxColumns, maxLists, maxListItems, maxLinks, maxJsonLd }) => { const clean = (v, n = 4000) => { const text = String(v || '').replace(/\s+/g, ' ').trim(); if (text.length > n) result.coverage.text_limit_hit = true; return text.slice(0, n); }; const result = { page_text: '', tables: [], lists: [], links: [], json_ld: [], coverage: { text_limit_hit: false, table_limit_hit: false, row_limit_hit: false, column_limit_hit: false, list_limit_hit: false, link_limit_hit: false, json_ld_limit_hit: false, virtualized_incomplete: false, virtualized_evidence: [], pagination_incomplete: false, pagination_evidence: [], scan_limit_hit: false, excluded_field_count: 0, excluded_hidden_count: 0, open_shadow_roots_visited: 0, scanned_node_count: 0 } }; // Read the composed DOM without cloning/mutating it or touching field values. const omittedFields = new WeakSet(); const omittedHidden = new WeakSet(); const shadowRoots = new WeakSet(); let visitedNodes = 0; const startedAt = performance.now(); const excludedTags = new Set(['INPUT','TEXTAREA','SELECT','OPTION','OUTPUT','SCRIPT','STYLE','TEMPLATE','NOSCRIPT','IFRAME','OBJECT','EMBED']); const fieldTags = new Set(['INPUT','TEXTAREA','SELECT','OPTION','OUTPUT']); const fieldRoles = new Set(['textbox','searchbox','combobox','spinbutton','slider']); const parentOf = el => el?.assignedSlot || el?.parentElement || el?.getRootNode?.()?.host || null; const fieldElement = el => fieldTags.has(el.tagName) || el.isContentEditable || fieldRoles.has((el.getAttribute('role') || '').toLowerCase()); const permitted = el => { let cursor = el; let depth = 0; while (cursor && depth++ < 96) { if (fieldElement(cursor)) { if (!omittedFields.has(cursor)) { omittedFields.add(cursor); result.coverage.excluded_field_count++; } return false; } if (excludedTags.has(cursor.tagName)) return false; // Range rectangles may exist for descendants of closed
. // Only its first summary participates in the rendered tree. if (cursor.tagName === 'DETAILS' && !cursor.open && cursor !== el) { const summary = Array.from(cursor.children).find(child => child.tagName === 'SUMMARY'); if (!summary || !(summary === el || summary.contains(el))) { if (!omittedHidden.has(cursor)) { omittedHidden.add(cursor); result.coverage.excluded_hidden_count++; } return false; } } const s = getComputedStyle(cursor); if (cursor.hidden || cursor.getAttribute('aria-hidden') === 'true' || cursor.hasAttribute('inert') || s.display === 'none' || s.visibility === 'hidden' || s.visibility === 'collapse' || Number(s.opacity) === 0) { if (!omittedHidden.has(cursor)) { omittedHidden.add(cursor); result.coverage.excluded_hidden_count++; } return false; } cursor = parentOf(cursor); } if (cursor) { result.coverage.scan_limit_hit = true; return false; } return true; }; const safeText = (root, limit = 4000, keepLines = false) => { if (!root || root.nodeType !== Node.ELEMENT_NODE || !permitted(root)) return ''; const pieces = []; const seen = new WeakSet(); let chars = 0; const push = text => { if (!text) return; if (chars + text.length > limit) result.coverage.text_limit_hit = true; const part = text.slice(0, Math.max(0, limit - chars)); pieces.push(part); chars += part.length; }; const walk = (node, depth) => { if (seen.has(node)) return; seen.add(node); visitedNodes++; if (visitedNodes > 30000 || depth > 96 || performance.now() - startedAt > 1800) { result.coverage.scan_limit_hit = true; return; } if (chars >= limit) { result.coverage.text_limit_hit = true; return; } if (node.nodeType === Node.TEXT_NODE) { const parent = node.parentElement || node.getRootNode()?.host; if (!parent || !permitted(parent)) return; const range = document.createRange(); range.selectNodeContents(node); const rendered = Array.from(range.getClientRects()).some(r => r.width > 0 && r.height > 0); if (rendered) push(String(node.nodeValue || '').replace(/\s+/g, ' ')); return; } if (node.nodeType !== Node.ELEMENT_NODE) return; if (fieldElement(node)) { if (!omittedFields.has(node)) { omittedFields.add(node); result.coverage.excluded_field_count++; } // Internal boundary only. Removed after the unchanged auth filter. // It prevents an empty password field consuming the next paragraph. if (keepLines) push(' [FORM_VALUE_OMITTED] '); return; } if (!permitted(node)) return; const display = getComputedStyle(node).display; const block = node.tagName === 'BR' || node.tagName === 'LABEL' || /^(block|flex|grid|list-item|flow-root|table|table-row|table-cell|table-caption)$/.test(display); if (block) push('\n'); let children; if (node.tagName === 'DETAILS' && !node.open) { const summary = Array.from(node.children).find(child => child.tagName === 'SUMMARY'); children = summary ? [summary] : []; if (node.childNodes.length > children.length && !omittedHidden.has(node)) { omittedHidden.add(node); result.coverage.excluded_hidden_count++; } } else if (node.shadowRoot) { if (!shadowRoots.has(node.shadowRoot)) { shadowRoots.add(node.shadowRoot); result.coverage.open_shadow_roots_visited++; } children = node.shadowRoot.childNodes; } else if (node.tagName === 'SLOT' && node.assignedNodes().length) { children = node.assignedNodes({flatten: true}); } else { children = node.childNodes; } for (const child of children) { if (chars >= limit || result.coverage.scan_limit_hit) break; walk(child, depth + 1); } if (block) push('\n'); }; walk(root, 0); result.coverage.scanned_node_count = visitedNodes; const lines = pieces.join('').split(/\n/).map(s => s.replace(/\s+/g, ' ').trim()).filter(Boolean); return keepLines ? lines.join('\n') : lines.join(' '); }; if (mode === 'all' || mode === 'text') { result.page_text = safeText(document.body, 200000, true); } if (mode === 'all' || mode === 'tables') { const tableNodes = Array.from(document.querySelectorAll('table')).filter(permitted); if (tableNodes.length > maxTables) result.coverage.table_limit_hit = true; for (const [ti, table] of tableNodes.slice(0, maxTables).entries()) { const allRows = Array.from(table.rows || []).filter(permitted); if (allRows.length > maxRows) result.coverage.row_limit_hit = true; const domRows = allRows.slice(0, maxRows); if (!domRows.length) continue; const firstCellsAll = Array.from(domRows[0].cells || []); if (firstCellsAll.length > maxColumns) result.coverage.column_limit_hit = true; const firstCells = firstCellsAll.slice(0, maxColumns); const header = firstCells.some(c => c.tagName.toLowerCase() === 'th'); const headers = firstCells.map((c, i) => safeText(c, 512) || `column_${i+1}`); const rows = (header ? domRows.slice(1) : domRows).map(row => { const allCells = Array.from(row.cells || []); if (allCells.length > maxColumns) result.coverage.column_limit_hit = true; return allCells.slice(0, maxColumns).map(c => safeText(c, 512)); }).filter(row => row.some(Boolean)); const declared = (table.hasAttribute('aria-rowcount') ? Number(table.getAttribute('aria-rowcount')) : NaN); if (Number.isFinite(declared) && declared > rows.length + (header ? 1 : 0)) { result.coverage.virtualized_incomplete = true; result.coverage.virtualized_evidence.push({table_index: ti + 1, declared_row_count: declared, observed_dom_rows: rows.length}); } result.tables.push({table_index: ti + 1, table_kind: 'html_table', caption: safeText(table.caption, 500), headers, rows, row_count: rows.length, declared_row_count: Number.isFinite(declared) ? declared : null, column_count: Math.max(headers.length, ...rows.map(r => r.length), 0)}); } const grids = Array.from(document.querySelectorAll('[role="grid"],[role="treegrid"]')).filter(g => g.tagName.toLowerCase() !== 'table' && permitted(g)); if (result.tables.length + grids.length > maxTables) result.coverage.table_limit_hit = true; for (const grid of grids.slice(0, Math.max(0, maxTables - result.tables.length))) { const headerNodes = Array.from(grid.querySelectorAll('[role="columnheader"]')).slice(0, maxColumns); const headers = headerNodes.map((c, i) => safeText(c, 512) || `column_${i+1}`); const allRows = Array.from(grid.querySelectorAll('[role="row"]')).filter(permitted); if (allRows.length > maxRows) result.coverage.row_limit_hit = true; const rows = []; for (const row of allRows.slice(0, maxRows)) { const cellsAll = Array.from(row.querySelectorAll('[role="gridcell"],[role="cell"]')); if (!cellsAll.length) continue; if (cellsAll.length > maxColumns) result.coverage.column_limit_hit = true; rows.push(cellsAll.slice(0, maxColumns).map(c => safeText(c, 512))); } const declared = (grid.hasAttribute('aria-rowcount') ? Number(grid.getAttribute('aria-rowcount')) : NaN); if (Number.isFinite(declared) && declared > rows.length) { result.coverage.virtualized_incomplete = true; result.coverage.virtualized_evidence.push({table_index: result.tables.length + 1, declared_row_count: declared, observed_dom_rows: rows.length}); } result.tables.push({table_index: result.tables.length + 1, table_kind: 'aria_grid', caption: clean(grid.getAttribute('aria-label') || '', 500), headers, rows, row_count: rows.length, declared_row_count: Number.isFinite(declared) ? declared : null, column_count: Math.max(headers.length, ...rows.map(r => r.length), 0)}); } } if (mode === 'all' || mode === 'lists') { const roots = Array.from(document.querySelectorAll('ul,ol')).filter(x => permitted(x) && (!x.parentElement || !x.parentElement.closest('ul,ol'))); if (roots.length > maxLists) result.coverage.list_limit_hit = true; for (const [li, list] of roots.slice(0, maxLists).entries()) { const allItems = Array.from(list.children || []).filter(x => x.tagName?.toLowerCase() === 'li' && permitted(x)); if (allItems.length > maxListItems) result.coverage.list_limit_hit = true; const items = allItems.slice(0, maxListItems).map(x => safeText(x, 4000)).filter(Boolean); if (items.length) result.lists.push({list_index: li + 1, ordered: list.tagName.toLowerCase() === 'ol', items}); } } if (mode === 'all' || mode === 'links') { const allLinks = Array.from(document.querySelectorAll('a[href]')).filter(a => { const s = getComputedStyle(a); const r = a.getBoundingClientRect(); return permitted(a) && s.visibility !== 'hidden' && s.display !== 'none' && r.width > 0 && r.height > 0; }); if (allLinks.length > maxLinks) result.coverage.link_limit_hit = true; result.links = allLinks.slice(0, maxLinks).map(a => ({text: safeText(a, 1000) || clean(a.getAttribute('aria-label'), 1000), url: clean(a.href, 3000)})); } if (mode === 'all' || mode === 'json_ld') { const scripts = Array.from(document.querySelectorAll('script[type="application/ld+json"]')); if (scripts.length > maxJsonLd) result.coverage.json_ld_limit_hit = true; for (const node of scripts.slice(0, maxJsonLd)) { const fullRaw = String(node.textContent || '').trim(); if (fullRaw.length > 200000) result.coverage.text_limit_hit = true; const raw = fullRaw.slice(0, 200000); if (!raw) continue; try { result.json_ld.push(JSON.parse(raw)); } catch { result.json_ld.push({_parse_error: true, raw}); } } } const nextSignals = Array.from(document.querySelectorAll('a[rel="next"],button,a')).filter(el => { const label = clean(el.getAttribute('aria-label') || el.getAttribute('title') || safeText(el, 120), 120); return el.matches('a[rel="next"]') || /^(next|next page|older|more|tiếp|trang sau|sau)$/i.test(label); }).slice(0, 8).map(el => clean(el.getAttribute('aria-label') || el.getAttribute('title') || safeText(el, 120), 120)); if (nextSignals.length) { result.coverage.pagination_incomplete = true; result.coverage.pagination_evidence = nextSignals; } return result; } """ try: return await frame.evaluate( script, { "mode": mode, "maxTables": MAX_TABLES, "maxRows": MAX_ROWS_PER_TABLE, "maxColumns": MAX_COLUMNS, "maxLists": MAX_LISTS, "maxListItems": MAX_LIST_ITEMS, "maxLinks": MAX_LINKS, "maxJsonLd": MAX_JSON_LD, }, ) except Exception as exc: raise BrowserBridgeError( "READ_DATA_EXTRACTION_FAILED", "The selected page/frame could not be read semantically.", details={"error_type": type(exc).__name__}, ) from exc @staticmethod def _coverage(extracted: dict[str, Any], records: list[dict[str, Any]]) -> dict[str, Any]: raw = dict(extracted.get("coverage") or {}) reasons = [] for key in ("text_limit_hit", "table_limit_hit", "row_limit_hit", "column_limit_hit", "list_limit_hit", "link_limit_hit", "json_ld_limit_hit", "scan_limit_hit"): if raw.get(key): reasons.append(key) if raw.get("virtualized_incomplete"): reasons.append("virtualized_rows_not_fully_present_in_current_dom") if raw.get("pagination_incomplete"): reasons.append("next_page_signal_present_but_read_data_does_not_navigate") if any(item.get("record_truncated") for item in records): reasons.append("normalized_record_output_bound") return { "scope": "selected_page_selected_frame_current_dom_only", "reader_revision": VERSION, "field_values_included": False, "excluded_field_count": int(raw.get("excluded_field_count", 0)), "excluded_hidden_count": int(raw.get("excluded_hidden_count", 0)), "scanned_node_count": int(raw.get("scanned_node_count", 0)), "open_shadow_roots_visited": int(raw.get("open_shadow_roots_visited", 0)), "shadow_scope": "text_open_composed_roots_structured_modes_light_dom_closed_unverified", "current_dom_complete_within_bounds": not any( raw.get(key) for key in ("text_limit_hit", "table_limit_hit", "row_limit_hit", "column_limit_hit", "list_limit_hit", "link_limit_hit", "json_ld_limit_hit", "scan_limit_hit") ) and not any(item.get("record_truncated") for item in records), "logical_dataset_complete": False if raw.get("virtualized_incomplete") or raw.get("pagination_incomplete") else None, "virtualized_data_complete": False if raw.get("virtualized_incomplete") else None, "pagination_complete": False if raw.get("pagination_incomplete") else None, "incomplete_reasons": list(dict.fromkeys(reasons)), "virtualized_evidence": list(raw.get("virtualized_evidence") or [])[:20], "pagination_evidence": list(raw.get("pagination_evidence") or [])[:8], "absence_of_incomplete_signal_is_not_proof_of_full_logical_dataset": True, } async def read_data( self, owner: str, *, session_id: str, lease_id: str, page_id: str = "", frame_id: str = "", mode: str = "tables", cursor: str = "", max_output_chars: int = DEFAULT_OUTPUT_CHARS, ) -> dict[str, Any]: selected_mode = str(mode or "tables").casefold().strip() if selected_mode not in {"all", "tables", "lists", "links", "json_ld", "text"}: raise BrowserBridgeError("READ_DATA_MODE_INVALID", "mode must be all, tables, lists, links, json_ld, or text.") if type(max_output_chars) is not int or not MIN_OUTPUT_CHARS <= max_output_chars <= MAX_OUTPUT_CHARS: raise BrowserBridgeError("READ_DATA_OUTPUT_BOUND_INVALID", f"max_output_chars must be {MIN_OUTPUT_CHARS} to {MAX_OUTPUT_CHARS}.") if not str(lease_id or "").startswith("lease_"): raise BrowserBridgeError("READ_DATA_LEASE_REQUIRED", "An exact owned-session lease_id is required.") # manager.get() revalidates owner/session/lease/TTL/current origin because app.py wraps this # operation in manager.lease_scope(lease_id). Continuations therefore cannot bypass lease expiry. session = await self.manager.get(owner, session_id, touch=False) if getattr(session, "lease", None) is None: raise BrowserBridgeError("READ_DATA_LEASE_REQUIRED", "Semantic read_data requires an owned leased session.") if cursor: async with session.lock: await self.manager.revalidate_locked_session(owner, session_id, session) page = await self.cursors.continue_page( cursor_id=cursor, owner=owner, session_id=session_id, lease_id=lease_id, max_output_chars=max_output_chars, ) metadata = dict(page.pop("metadata")) return { "ok": True, "status": "READ_DATA_COMPLETE" if page.get("next_cursor") is None else "READ_DATA_PARTIAL", "session_id": session_id, "lease_required": True, "page_id": metadata["page_id"], "frame_id": metadata["frame_id"], "mode": metadata["mode"], "items": page["items"], "returned_count": page["returned_count"], "total_record_count": page["total_record_count"], "snapshot_id": page["snapshot_id"], "source_fingerprint": page["source_fingerprint"], "next_cursor": page["next_cursor"], "cursor_expires_in_ms": page["cursor_expires_in_ms"], "reader_revision": VERSION, "coverage": metadata["coverage"], "auth_field_omissions": metadata["auth_field_omissions"], "page_inventory": metadata["page_inventory"], "frame_inventory": metadata["frame_inventory"], "blocked_page_count": metadata["blocked_page_count"], "blocked_frame_count": metadata["blocked_frame_count"], "attachment_returned": False, "page_or_frame_changed": False, "content_origin": "untrusted_web_page", "secret_values_exposed": False, } async with session.lock: await self.manager.revalidate_locked_session(owner, session_id, session) pages, page_map, blocked_page_count = self._page_inventory(session) if not pages: raise BrowserBridgeError( "READ_DATA_NO_LEASE_SCOPED_PAGES", "The owned session has no readable page inside the task lease origin set.", details={"blocked_page_count": blocked_page_count}, ) if page_id: if not _PAGE_RE.fullmatch(page_id) or page_id not in page_map: raise BrowserBridgeError("READ_DATA_PAGE_NOT_FOUND", "The requested page_id is unavailable. Read the returned page inventory again.") page = page_map[page_id] selected_page_id = page_id elif len(pages) == 1: selected_page_id = pages[0]["page_id"] page = page_map[selected_page_id] else: return { "ok": True, "status": "READ_DATA_PAGE_SELECTION_REQUIRED", "session_id": session_id, "lease_required": True, "page_inventory": pages, "page_count": len(pages), "blocked_page_count": blocked_page_count, "complete": False, "attachment_returned": False, "secret_values_exposed": False, } frames, frame_map, blocked_frame_count = self._frame_inventory(session, page) if frame_id: if not _FRAME_RE.fullmatch(frame_id) or frame_id not in frame_map: raise BrowserBridgeError("READ_DATA_FRAME_NOT_FOUND", "The requested frame_id is unavailable. Read the returned frame inventory again.") frame = frame_map[frame_id] selected_frame_id = frame_id else: frame = page.main_frame selected_frame_id = self._stable_frame_id(session_id, frame) self._require_frame_origin(session, frame) extracted = await self._extract_frame(frame, selected_mode) # Keep a standalone auth label with the following block before filtering. # This avoids exposing a value laid out on the next line after "Password:". lines = str(extracted.get("page_text", "")).splitlines() merged = [] index = 0 while index < len(lines): line = lines[index] if _AUTH_LINE_RE.search(line) and "[FORM_VALUE_OMITTED]" not in line and (line.endswith(":") or len(line.split()) <= 3) and index + 1 < len(lines): line += " " + lines[index + 1] index += 1 merged.append(line) index += 1 extracted["page_text"] = "\n".join(merged) sanitized, omitted = self._sanitize_value(extracted) if isinstance(sanitized, dict): sanitized["page_text"] = str(sanitized.get("page_text", "")).replace("[FORM_VALUE_OMITTED]", "") sanitized = sanitized if isinstance(sanitized, dict) else {} records = self._records(sanitized) coverage = self._coverage(sanitized, records) if omitted or coverage["excluded_field_count"]: coverage["returned_content_complete"] = False reasons = coverage["incomplete_reasons"] if omitted: reasons.append("auth_content_redacted") if coverage["excluded_field_count"]: reasons.append("form_value_subtrees_excluded") page_origin, page_path = inventory_location(page.url) frame_origin, frame_path = inventory_location(frame.url) fingerprint_payload = { "session_id": session_id, "page_id": selected_page_id, "frame_id": selected_frame_id, "mode": selected_mode, "records": records, "coverage": coverage, } fingerprint = hashlib.sha256(json.dumps(fingerprint_payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() metadata = { "page_id": selected_page_id, "frame_id": selected_frame_id, "mode": selected_mode, "coverage": coverage, "auth_field_omissions": omitted, "page_inventory": pages, "frame_inventory": frames, "blocked_page_count": blocked_page_count, "blocked_frame_count": blocked_frame_count, "page_origin": page_origin, "page_path": page_path, "frame_origin": frame_origin, "frame_path": frame_path, } frozen = await self.cursors.start( owner=owner, session_id=session_id, lease_id=lease_id, page_id=selected_page_id, frame_id=selected_frame_id, mode=selected_mode, max_output_chars=max_output_chars, records=records, metadata=metadata, source_fingerprint=fingerprint, ) metadata = dict(frozen.pop("metadata")) return { "ok": True, "status": "READ_DATA_COMPLETE" if frozen.get("next_cursor") is None else "READ_DATA_PARTIAL", "session_id": session_id, "lease_required": True, "page_id": selected_page_id, "frame_id": selected_frame_id, "page_origin": metadata["page_origin"], "page_path": metadata["page_path"], "frame_origin": metadata["frame_origin"], "frame_path": metadata["frame_path"], "mode": selected_mode, "items": frozen["items"], "returned_count": frozen["returned_count"], "total_record_count": frozen["total_record_count"], "snapshot_id": frozen["snapshot_id"], "source_fingerprint": frozen["source_fingerprint"], "next_cursor": frozen["next_cursor"], "cursor_expires_in_ms": frozen["cursor_expires_in_ms"], "reader_revision": VERSION, "coverage": coverage, "auth_field_omissions": omitted, "page_inventory": pages, "frame_inventory": frames, "blocked_page_count": blocked_page_count, "blocked_frame_count": blocked_frame_count, "attachment_returned": False, "page_or_frame_changed": False, "screenshot_taken": False, "os_input_used": False, "content_origin": "untrusted_web_page", "secret_values_exposed": False, }