diff --git a/docs/config.md b/docs/config.md index 440b4e2..2cd3428 100644 --- a/docs/config.md +++ b/docs/config.md @@ -598,6 +598,42 @@ an operator's separately selected outer deadline still applies. `replay` retains the original observation metadata and is unchanged; it does not apply aliases or silently correct historical labels. +Inspect retained family menus and variable-selection forms before composing the +next explicit request: + +```bash +uv run python -m teadata.scripts.refresh_tapr form \ + --source-store provenance --record FORM_ACQUISITION_ID \ + --encoding iso-8859-1 --store provenance --output form-observations.json +``` + +This offline inspection records source-order control and option attributes, +labels, literal method/action values, tag coordinates, and fieldset/optgroup +context. A shared context table uses parent indices; controls reference that +table instead of copying their ancestors. It preserves repeated control names +and empty values. Missing methods +stay unknown. A missing form end tag is reported as `closed: false`; ambiguous +nested forms, controls or duplicate attributes fail for source review. Display +labels normalize presentation whitespace; textarea content stays exact. +The typed `teadata.tapr_forms.extract_tapr_forms()` API provides the same snapshots. + +The report resolves anchor URLs and literal `window.open` first arguments +against the retained acquisition's observed URL. It never executes JavaScript, +evaluates disabled/checked/select submission rules, infers selected variables, +submits a form, or downloads data. Computed event handlers remain literal +attributes for review. A local observation without a recorded HTTP URL keeps +relative targets unresolved. The original bytes and acquisition dates remain +authoritative; a new private processing receipt records this inspection. +Inputs use the same explicit positive `--max-bytes` bound and SHA/length checks +as dictionary inspection, and existing output paths are never overwritten. +`--max-nodes` bounds the complete form projection (default 20,000 observed forms, +controls, options, labels, links and context nodes). `--max-output-bytes` bounds +streamed JSON staging for both form and dictionary inspection (default 8 MiB). +Oversized output is discarded before publishing a processing receipt or report; +larger reviewed inspections require explicit limits. Retained inputs must be +regular files, are opened without following a final symlink or blocking on a +FIFO, and must retain the same file identity throughout capture. + Inspect a retained TAPR HTML dictionary offline through the same command owner: ```bash diff --git a/teadata/scripts/refresh_tapr.py b/teadata/scripts/refresh_tapr.py index a41b61c..5efe530 100644 --- a/teadata/scripts/refresh_tapr.py +++ b/teadata/scripts/refresh_tapr.py @@ -12,6 +12,7 @@ import json import os from pathlib import Path +import stat import sys import tempfile from urllib.error import HTTPError @@ -23,6 +24,8 @@ from teadata.provenance import ProvenanceStore, _identifier from teadata.tapr import compare_tapr_tables, read_tapr_table +INSPECTION_OUTPUT_MAX_BYTES = 8 * 1024 * 1024 + def _read_pinned_input(path: Path, expected_sha256: str) -> bytes: """Capture one small explicit input; the same bytes are parsed and retained.""" @@ -219,61 +222,122 @@ def compare_inputs(args) -> dict: 'report_sha256': sha256(output.read_bytes()).hexdigest()} -def inspect_dictionary(args) -> dict: - """Inspect one retained HTML artifact offline, preserving its exact history.""" - from teadata.tapr_dictionary import extract_tapr_dictionary - +def _inspection_input(args, label: str) -> tuple[ProvenanceStore, dict, bytes]: + """Capture the exact bounded HTML bytes whose acquisition is retained.""" if args.output.exists() or args.output.is_symlink(): - raise FileExistsError("Dictionary output already exists; select a new observation path") + raise FileExistsError(f"{label} output already exists; select a new observation path") if type(args.max_bytes) is not int or args.max_bytes < 1: - raise ValueError('Dictionary max-bytes must be a positive integer') + raise ValueError(f'{label} max-bytes must be a positive integer') + output_limit = getattr(args, 'max_output_bytes', INSPECTION_OUTPUT_MAX_BYTES) + if type(output_limit) is not int or output_limit < 1: + raise ValueError('Inspection max-output-bytes must be a positive integer') source, archive = ProvenanceStore(args.source_store), ProvenanceStore(args.store) record = source.read_record(args.record) if record['kind'] != 'acquisition': - raise ValueError('Dictionary input must be a retained acquisition observation') + raise ValueError(f'{label} input must be a retained acquisition observation') artifact = record['artifact'] expected_size = artifact['size_bytes'] if type(expected_size) is not int or expected_size < 0: - raise ValueError('Invalid retained dictionary size') + raise ValueError(f'Invalid retained {label.lower()} size') if expected_size > args.max_bytes: - raise ValueError('Retained dictionary exceeds max-bytes') + raise ValueError(f'Retained {label.lower()} exceeds max-bytes') selected = source.root / 'objects' / _identifier(artifact['sha256']) - with selected.open('rb') as stream: - if os.fstat(stream.fileno()).st_size > args.max_bytes: - raise ValueError('Retained dictionary exceeds max-bytes') + fd = os.open(selected, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(fd, 'rb') as stream: + before = os.fstat(stream.fileno()) + if not stat.S_ISREG(before.st_mode): + raise ValueError('Retained inspection input must be a regular file') + if before.st_size > args.max_bytes: + raise ValueError(f'Retained {label.lower()} exceeds max-bytes') raw = stream.read(args.max_bytes + 1) + after = os.fstat(stream.fileno()) + fields = ('st_dev', 'st_ino', 'st_mode', 'st_size', 'st_mtime_ns', 'st_ctime_ns') + current = selected.stat(follow_symlinks=False) + if any(getattr(before, key) != getattr(after, key) or getattr(after, key) != getattr(current, key) + for key in fields): + raise ValueError('Retained inspection input changed during capture') if len(raw) > args.max_bytes: - raise ValueError('Retained dictionary exceeds max-bytes') + raise ValueError(f'Retained {label.lower()} exceeds max-bytes') if len(raw) != expected_size or sha256(raw).hexdigest() != artifact['sha256']: - raise ValueError('Parsed dictionary bytes differ from the selected acquisition') + raise ValueError(f'Parsed {label.lower()} bytes differ from the selected acquisition') archive.copy_record_from(source, args.record) + return archive, record, raw + + +def _save_inspection(args, archive: ProvenanceStore, report: dict, *, kind: str, + implementation: dict, validation: dict) -> dict: + """Retain one new inspection and its processing closure without overwriting.""" + report['implementation_sha256'] = implementation + args.output.parent.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix=f'.tapr-{kind}-', dir=args.output.parent) as directory: + staged = Path(directory) / args.output.name + output_limit = getattr(args, 'max_output_bytes', INSPECTION_OUTPUT_MAX_BYTES) + written = 0 + with staged.open('xb') as stream: + os.fchmod(stream.fileno(), 0o600) + for text in json.JSONEncoder(indent=2, ensure_ascii=True, allow_nan=False).iterencode(report): + chunk = text.encode('utf-8') + written += len(chunk) + if written + 1 > output_limit: + raise ValueError('Inspection output exceeds max-output-bytes') + stream.write(chunk) + stream.write(b'\n') + processing = archive.record_processing(dataset=f'tapr_explicit_{kind}_inspection', inputs=[args.record], + outputs={f'{kind}_occurrences': staged}, processor='teadata.scripts.refresh_tapr', + code_version=sha256(json.dumps(implementation, sort_keys=True).encode()).hexdigest(), + parameters={'encoding': args.encoding, 'max_bytes': args.max_bytes, + 'max_output_bytes': output_limit, 'no_config_activation': True, + **({'max_nodes': report['max_nodes']} if kind == 'form' else {})}, + validation=validation) + archive.verify_record_closure([processing]) + os.link(staged, args.output) + return {'status': f'inspected_{kind}', 'processing_record_id': processing, + 'report_sha256': sha256(args.output.read_bytes()).hexdigest()} + + +def inspect_dictionary(args) -> dict: + """Inspect one retained HTML dictionary offline, preserving its history.""" + from teadata.tapr_dictionary import extract_tapr_dictionary + + archive, record, raw = _inspection_input(args, 'Dictionary') occurrences = extract_tapr_dictionary(raw, encoding=args.encoding) root = Path(__file__).resolve().parents[1] implementation = {name: sha256((root / name).read_bytes()).hexdigest() for name in ['tapr_dictionary.py', 'scripts/refresh_tapr.py']} report = {'schema_version': 1, 'source_record_id': args.record, - 'source_artifact': artifact, 'encoding': args.encoding, 'max_bytes': args.max_bytes, + 'source_artifact': record['artifact'], 'encoding': args.encoding, 'max_bytes': args.max_bytes, 'coordinate_semantics': 'Outer tooltip start tag; one-based line, zero-based decoded-character column', 'text_projection': 'Decoded rendered text; presentation whitespace only is normalized. Raw artifact owns original bytes.', - 'dictionary_records': [row.column_record() for row in occurrences], - 'implementation_sha256': implementation, 'source_activation': False} - args.output.parent.mkdir(parents=True, exist_ok=True) - with tempfile.TemporaryDirectory(prefix='.tapr-dictionary-', dir=args.output.parent) as directory: - staged = Path(directory) / args.output.name - staged.write_text(json.dumps(report, indent=2, ensure_ascii=True, allow_nan=False) + '\n') - staged.chmod(0o600) - processing = archive.record_processing(dataset='tapr_explicit_dictionary_inspection', inputs=[args.record], - outputs={'dictionary_occurrences': staged}, processor='teadata.scripts.refresh_tapr', - code_version=sha256(json.dumps(implementation, sort_keys=True).encode()).hexdigest(), - parameters={'encoding': args.encoding, 'max_bytes': args.max_bytes, 'no_config_activation': True}, - validation={'occurrences': len(occurrences), 'scientific_equivalence_claimed': False}) - archive.verify_record_closure([processing]) - os.link(staged, args.output) - return {'status': 'inspected_dictionary', 'processing_record_id': processing, - 'report_sha256': sha256(args.output.read_bytes()).hexdigest()} + 'dictionary_records': [row.column_record() for row in occurrences], 'source_activation': False} + return _save_inspection(args, archive, report, kind='dictionary', implementation=implementation, + validation={'occurrences': len(occurrences), 'scientific_equivalence_claimed': False}) + + +def inspect_form(args) -> dict: + """Observe retained controls and literal links; never submit or evaluate JS.""" + from teadata.tapr_forms import DEFAULT_FORM_NODES, extract_tapr_forms + + archive, record, raw = _inspection_input(args, 'Form') + base_url = record.get('resolved_url') or record.get('requested_url') + max_nodes = getattr(args, 'max_nodes', DEFAULT_FORM_NODES) + forms = extract_tapr_forms(raw, encoding=args.encoding, base_url=base_url, max_nodes=max_nodes) + root = Path(__file__).resolve().parents[1] + implementation = {name: sha256((root / name).read_bytes()).hexdigest() + for name in ['tapr_forms.py', 'scripts/refresh_tapr.py']} + report = {'schema_version': 1, 'source_record_id': args.record, 'source_artifact': record['artifact'], + 'encoding': args.encoding, 'max_bytes': args.max_bytes, 'max_nodes': max_nodes, 'base_url': base_url, + 'coordinate_semantics': 'Start tag; one-based line, zero-based decoded-character column', + 'text_projection': 'Display-label whitespace normalized; textarea text exact. Raw artifact owns original bytes.', + 'forms': [form.to_dict() for form in forms], 'source_activation': False, + 'submission_controls_inferred': False, 'javascript_executed': False} + return _save_inspection(args, archive, report, kind='form', implementation=implementation, + validation={'forms': len(forms), 'closed_forms': sum(form.closed for form in forms), + 'scientific_equivalence_claimed': False, 'data_downloaded': False}) def main(argv: list[str] | None = None) -> int: + from teadata.tapr_forms import DEFAULT_FORM_NODES + parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest='command', required=True) acquire = commands.add_parser('acquire', help='Explicit bounded public GET/POST acquisition; never change Config') @@ -307,6 +371,16 @@ def main(argv: list[str] | None = None) -> int: help='Positive retained HTML byte limit; default 2 MiB') dictionary.add_argument('--store', type=Path, required=True, help='Append-only output provenance archive') dictionary.add_argument('--output', type=Path, required=True, help='New occurrence report; never overwritten') + dictionary.add_argument('--max-output-bytes', type=int, default=INSPECTION_OUTPUT_MAX_BYTES) + form = commands.add_parser('form', help='Inspect retained HTML controls and literal links offline') + form.add_argument('--source-store', type=Path, required=True) + form.add_argument('--record', required=True) + form.add_argument('--encoding', required=True) + form.add_argument('--max-bytes', type=int, default=2 * 1024 * 1024) + form.add_argument('--max-output-bytes', type=int, default=INSPECTION_OUTPUT_MAX_BYTES) + form.add_argument('--max-nodes', type=int, default=DEFAULT_FORM_NODES) + form.add_argument('--store', type=Path, required=True, help='Append-only inspection provenance archive') + form.add_argument('--output', type=Path, required=True, help='New form report; never overwritten') compare = commands.add_parser('compare', help='Compare two pinned reviewed original inputs offline') compare.add_argument('--store', type=Path, required=True, help='Append-only report/provenance archive') compare.add_argument('--output', type=Path, required=True, help='New report path; never overwritten') @@ -325,6 +399,8 @@ def main(argv: list[str] | None = None) -> int: result = compare_inputs(args) elif args.command == 'dictionary': result = inspect_dictionary(args) + elif args.command == 'form': + result = inspect_form(args) else: archive = ProvenanceStore(args.store) if args.command == 'replay': @@ -341,7 +417,7 @@ def main(argv: list[str] | None = None) -> int: record = archive.read_record(acquired.record_id) result = {'status': 'acquired', 'record_id': acquired.record_id, 'artifact': record['artifact']} print(json.dumps(result, sort_keys=True)) - return 0 if result['status'] in {'compared', 'acquired', 'inspected_dictionary'} else 2 + return 0 if result['status'] in {'compared', 'acquired', 'inspected_dictionary', 'inspected_form'} else 2 if __name__ == '__main__': diff --git a/teadata/tapr_forms.py b/teadata/tapr_forms.py new file mode 100644 index 0000000..4d6aa66 --- /dev/null +++ b/teadata/tapr_forms.py @@ -0,0 +1,243 @@ +"""Observe retained TAPR forms without submitting controls or executing scripts.""" +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field +from html.parser import HTMLParser +import re +from typing import Any +from urllib.parse import urljoin, urlsplit + +Attributes = tuple[tuple[str, str | None], ...] +DEFAULT_FORM_NODES = 20_000 +_WINDOW_OPEN = re.compile(r"\bwindow\.open\(\s*(['\"])([^'\"\\]+)\1\s*[,)]") + + +def _attributes(items: list[tuple[str, str | None]]) -> Attributes: + names = [name for name, _ in items] + if len(names) != len(set(names)): + raise ValueError("Ambiguous duplicate form attribute") + return tuple(items) + + +def _text(parts: list[str]) -> str: + return " ".join("".join(parts).split()) + + +def _resolve(base_url: str | None, value: str | None) -> str | None: + if value is None or base_url is None: + return None + resolved = urljoin(base_url, value) + return resolved if urlsplit(resolved).scheme in {"http", "https"} else None + + +@dataclass(frozen=True) +class FormSnapshot: + """Detached source-order facts; these are not successful submission controls.""" + + attributes: Attributes + position: tuple[int, int] + controls: tuple[dict[str, Any], ...] + labels: tuple[dict[str, Any], ...] + links: tuple[dict[str, Any], ...] + contexts: tuple[dict[str, Any], ...] + closed: bool + base_url: str | None + + def to_dict(self) -> dict[str, Any]: + attributes = dict(self.attributes) + return {"attributes": list(self.attributes), "position": list(self.position), + "method": attributes.get("method"), "action": attributes.get("action"), + "resolved_action": _resolve(self.base_url, attributes.get("action")), + "controls": deepcopy(list(self.controls)), "labels": deepcopy(list(self.labels)), + "links": deepcopy(list(self.links)), "contexts": deepcopy(list(self.contexts)), + "closed": self.closed} + + +@dataclass +class _Form: + attributes: Attributes + position: tuple[int, int] + controls: list[dict[str, Any]] = field(default_factory=list) + labels: list[dict[str, Any]] = field(default_factory=list) + links: list[dict[str, Any]] = field(default_factory=list) + contexts: list[dict[str, Any]] = field(default_factory=list) + + +class _Forms(HTMLParser): + def __init__(self, base_url: str | None, max_nodes: int): + super().__init__(convert_charrefs=True) + self.base_url = base_url + self.forms: list[FormSnapshot] = [] + self.form: _Form | None = None + self.control: dict[str, Any] | None = None + self.control_text: list[str] = [] + self.option: dict[str, Any] | None = None + self.option_text: list[str] = [] + self.label: dict[str, Any] | None = None + self.label_text: list[str] = [] + self.ignored: list[str] = [] + self.context: list[int] = [] + self.nodes, self.max_nodes = 0, max_nodes + + def take_node(self) -> None: + self.nodes += 1 + if self.nodes > self.max_nodes: + raise ValueError("Form projection exceeds max-nodes") + + def finish_label(self) -> None: + if self.label is not None and self.form is not None: + self.label["text"] = _text(self.label_text) + self.form.labels.append(self.label) + self.label, self.label_text = None, [] + + def finish_option(self) -> None: + if self.option is not None and self.control is not None: + self.option["text"] = _text(self.option_text) + self.control["options"].append(self.option) + self.option, self.option_text = None, [] + + def finish_control(self) -> None: + self.finish_option() + if self.control is not None: + self.control["text"] = ("".join(self.control_text) if self.control["tag"] == "textarea" + else _text(self.control_text)) + if (self.control["tag"] == "select" and self.context and self.form is not None + and self.form.contexts[self.context[-1]]["tag"] == "optgroup"): + self.context.pop() + self.control, self.control_text = None, [] + + def finish_form(self, *, closed: bool) -> None: + if self.form is None: + raise ValueError("Unmatched form end tag") + self.finish_label() + self.finish_control() + form = self.form + self.forms.append(FormSnapshot(form.attributes, form.position, tuple(form.controls), + tuple(form.labels), tuple(form.links), tuple(form.contexts), closed, self.base_url)) + self.form = None + self.context = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + if tag in {"script", "style"}: + self.ignored.append(tag) + return + if self.ignored: + return + if tag == "form": + if self.form is not None: + raise ValueError("Nested forms require source review") + self.take_node() + self.form = _Form(_attributes(attrs), self.getpos()) + if self.form is None: + return + if tag in {"fieldset", "legend", "optgroup"}: + if tag == "optgroup": + if self.control is None or self.control["tag"] != "select": + raise ValueError("Option group requires a select control") + self.finish_option() + if self.context and self.form.contexts[self.context[-1]]["tag"] == "optgroup": + self.context.pop() # The preceding optgroup end tag is optional. + elif self.control is not None and self.control["tag"] == "select": + raise ValueError("Unsupported fieldset/legend context inside select") + self.take_node() + self.form.contexts.append({"tag": tag, "attributes": list(_attributes(attrs)), + "position": list(self.getpos()), "parent": self.context[-1] if self.context else None}) + self.context.append(len(self.form.contexts) - 1) + if tag in {"input", "select", "textarea", "button"}: + if self.control is not None: + raise ValueError("Nested form controls require source review") + self.take_node() + attributes = _attributes(attrs) + row: dict[str, Any] = {"tag": tag, "attributes": list(attributes), + "position": list(self.getpos()), + "context_id": self.context[-1] if self.context else None} + self.form.controls.append(row) + if tag != "input": + self.control = row + if tag == "select": + row["options"] = [] + for name, value in attributes: + if name == "onclick" and value: + for match in _WINDOW_OPEN.finditer(value): + self.take_node() + target = match[2] + self.form.links.append({"kind": "window.open_literal", "target": target, + "resolved_url": _resolve(self.base_url, target), "position": list(self.getpos())}) + elif tag == "option" and self.control is not None and self.control["tag"] == "select": + self.finish_option() + self.take_node() + self.option = {"attributes": list(_attributes(attrs)), "position": list(self.getpos()), + "context_id": self.context[-1] if self.context else None} + elif tag == "label": + # Publisher pages sometimes open another