diff --git a/docs/changelog/1197.bugfix.rst b/docs/changelog/1197.bugfix.rst new file mode 100644 index 00000000..69d6cb96 --- /dev/null +++ b/docs/changelog/1197.bugfix.rst @@ -0,0 +1 @@ +Exercise qualified public APIs through checked Atheris consumers. diff --git a/tests/atheris/test_public_targets.py b/tests/atheris/test_public_targets.py new file mode 100644 index 00000000..5dd28668 --- /dev/null +++ b/tests/atheris/test_public_targets.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import importlib +import importlib.util +import json +import os +import sys +from pathlib import Path +from subprocess import run # ruff: ignore[suspicious-subprocess-import] - libFuzzer owns a separate interpreter. +from typing import TYPE_CHECKING, Final, cast + +import pytest +from fuzz.atheris_runtime import build_runtime + +if TYPE_CHECKING: + from typing import Protocol + + class _Runtime(Protocol): + def path(self) -> str: ... + + +_CONSUME: Final = """ +import sys +from pathlib import Path +from typing import Final +from fuzz.atheris_targets import public_targets +_TARGET: Final = next(target for target in public_targets() if target.name == sys.argv[1]) +_TARGET.callback(Path(sys.argv[2]).read_bytes()) +""" + + +_LIST: Final = """ +import json +from fuzz.atheris_targets import owner_inventory, public_targets +print(json.dumps({"owners": len(owner_inventory()), "targets": [target.name for target in public_targets()]})) +""" + + +@pytest.mark.oracle +def test_atheris_public_consumers_run_with_native_coverage(tmp_path: Path) -> None: + if sys.platform != "linux" or importlib.util.find_spec("atheris") is None: + pytest.skip("Atheris's native runtime requires its optional Linux wheel") + runtime: Final = cast("_Runtime", importlib.import_module("atheris")) + library: Final = build_runtime(Path(runtime.path()) / "libclang_rt.fuzzer_no_main.a", tmp_path / "runtime") + environment: Final = { + **os.environ, + "LD_PRELOAD": str(library), + "PYTHONPATH": str(Path(__file__).parents[2] / "tools"), + "GCOV_PREFIX": str(tmp_path / "inventory-gcda"), + } + inventory: Final = run( # ruff: ignore[subprocess-without-shell-equals-true] - fixed interpreter and owned modules. + [sys.executable, "-c", _LIST], + env=environment, + capture_output=True, + text=True, + check=True, + ) + targets: Final = json.loads(inventory.stdout) + assert (targets["owners"], len(targets["targets"])) == (209, 28) + outcomes: Final[dict[str, dict[str, int | str]]] = {} + for target in targets["targets"]: + corpus: Final = tmp_path / target + corpus.mkdir() + seed: Final = ( + b"one" if target == "xml-schema" else b"p.x" if target == "css-translate" else "水😀".encode() + ) + (corpus / "utf8").write_bytes(seed) + counters: Final = tmp_path / f"{target}-gcda" + run( # ruff: ignore[subprocess-without-shell-equals-true] - public callback and private native counters. + [sys.executable, "-c", _CONSUME, target, str(corpus / "utf8")], + env={**environment, "GCOV_PREFIX": str(counters)}, + capture_output=True, + check=True, + ) + result = run( # ruff: ignore[subprocess-without-shell-equals-true] - fixed interpreter and finite public corpus. + [ + sys.executable, + "-m", + "fuzz.atheris_targets", + "--target", + target, + "--corpus", + str(corpus), + "-atheris_runs=8", + "-seed=1", + "-max_len=64", + "-detect_leaks=0", + ], + env={**environment, "GCOV_PREFIX": str(tmp_path / f"{target}-gcda")}, + capture_output=True, + check=False, + ) + (tmp_path / f"{target}.log").write_bytes(result.stdout + result.stderr) + assert result.returncode == 0, (target, result.stdout + result.stderr) + assert b"ATHERIS_REJECTION_BRIDGE=1" in result.stderr + assert b"Done 8" in result.stderr + assert b"inline 8-bit counters" in result.stderr + assert b"PC tables" in result.stderr + assert b"cov:" in result.stderr + assert b"ft:" in result.stderr + assert b"Coverage symbols are being provided by a library other than libFuzzer" not in result.stderr + manifest: Final = json.loads(corpus.with_suffix(".json").read_text()) + assert (manifest["target"], bool(manifest["exports"])) == (target, True) + native_counters: Final = tuple(counters.rglob("*.gcda")) + assert native_counters, target + outcomes[target] = { + "exit": result.returncode, + "corpus": str(corpus), + "native_counter_files": len(native_counters), + } + (tmp_path / "public-targets.json").write_text(json.dumps(outcomes, indent=2) + "\n", encoding="utf-8") diff --git a/tests/test_fuzz_atheris_content_targets.py b/tests/test_fuzz_atheris_content_targets.py new file mode 100644 index 00000000..1f010354 --- /dev/null +++ b/tests/test_fuzz_atheris_content_targets.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Final + +import pytest +from fuzz.atheris_content_targets import ( + content_targets, + minifier_check, + minifier_observation, + sanitizer_check, + sanitizer_observation, + stdlib_observation, + url_check, +) + +from turbohtml.clean import Removed + +if TYPE_CHECKING: + from fuzz.atheris_registry import Target + +_TARGETS: Final = content_targets() + + +@pytest.mark.parametrize("target", _TARGETS, ids=lambda target: target.name) +@pytest.mark.parametrize("data", [b"", b"hello", bytes(range(256))], ids=["empty", "ascii", "byte-range"]) +def test_content_target_behaviors(target: Target, data: bytes) -> None: + target.callback(data) + + +def test_content_targets_have_unique_owners() -> None: + exports: Final = [export for target in _TARGETS for export in target.exports] + assert len(exports) == len(set(exports)) == 92 + + +def test_content_sanitizer_removes_attributes_and_comments() -> None: + assert sanitizer_observation(b"x") == ( + "value78value78", + [Removed("b", "title"), Removed("b", "onclick")], + ) + + +def test_content_minifiers_transform_source() -> None: + assert minifier_observation(b"x") == ( + "

value78

", + ".value78{color:red}", + "color:red", + 'const value="value78"', + ) + + +def test_content_stdlib_decodes_chunked_references() -> None: + assert stdlib_observation(b"x") == "value78&" + + +def test_content_sanitizer_rejects_wrong_renderer_result() -> None: + def render(data: bytes) -> tuple[str, list[Removed]]: + return data.decode(), [] + + with pytest.raises(AssertionError, match="Content API mismatch"): + sanitizer_check(b"wrong", render) + + +@pytest.mark.parametrize("kind", ["css-semantics", "css-fixpoint", "js-fixpoint"]) +def test_content_minifier_rejects_wrong_output(kind: str) -> None: + def wrong(source: str) -> str: + return "q{color:blue}" if kind == "css-semantics" else source + "x" + + with pytest.raises(AssertionError, match="Content API mismatch"): + minifier_check(b"x", js=wrong) if kind == "js-fixpoint" else minifier_check(b"x", css=wrong) + + +def test_content_url_rejects_stable_wrong_host() -> None: + with pytest.raises(AssertionError, match="Content API mismatch"): + url_check(b"x", lambda _source: "https://wrong.example/") diff --git a/tests/test_fuzz_atheris_dom_targets.py b/tests/test_fuzz_atheris_dom_targets.py new file mode 100644 index 00000000..acb1e807 --- /dev/null +++ b/tests/test_fuzz_atheris_dom_targets.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +from typing import Final + +import pytest +from fuzz.atheris_dom_targets import DomObservation, dom_observation, dom_targets + +_DOMAINS: Final = ( + "dom-construction", + "dom-traversal", + "dom-query", + "dom-mutation", + "dom-range", + "dom-shadow", + "dom-locations", + "dom-rewrite", + "dom-sax", + "dom-treebuild", +) + + +@pytest.mark.parametrize("domain", _DOMAINS) +@pytest.mark.parametrize( + "data", + [ + pytest.param(b"", id="empty"), + pytest.param(b"text", id="plain"), + pytest.param("é水😀".encode(), id="unicode"), + pytest.param(b"<&>\r\n\x00", id="escaped"), + ], +) +def test_dom_independent_contract(domain: str, data: bytes) -> None: + observation: Final = dom_observation(data, domain) + assert observation.actual == observation.expected + + +@pytest.mark.parametrize("domain", _DOMAINS) +def test_dom_registered_callback(domain: str) -> None: + target: Final = next(target for target in dom_targets() if target.name == domain) + assert target.exceptions == (UnicodeDecodeError,) + target.callback(b"text") + with pytest.raises(UnicodeDecodeError): + target.callback(b"\xff") + + +def test_dom_ownership() -> None: + targets: Final = dom_targets() + exports: Final = [export for target in targets for export in target.exports] + assert (tuple(target.name for target in targets), len(exports), len(set(exports))) == (_DOMAINS, 64, 64) + + +def test_dom_verification_rejects_changed_result() -> None: + with pytest.raises(AssertionError, match=r"actual.*expected"): + DomObservation("actual", "expected").verify() + + +def test_dom_verification_accepts_observed_result() -> None: + assert dom_observation(b"text", "dom-range").verify() is None + + +def test_dom_range_literal() -> None: + assert dom_observation(b"<&>", "dom-range").actual == repr(("

<&>

", 0, 1, 0, 1, False, "<&>")) + + +def test_dom_input_bound() -> None: + assert dom_observation(b"x" * 65, "dom-range") == dom_observation(b"x" * 64, "dom-range") + + +def test_dom_unknown_domain() -> None: + with pytest.raises(KeyError, match="unknown"): + dom_observation(b"text", "unknown") diff --git a/tests/test_fuzz_atheris_driver.py b/tests/test_fuzz_atheris_driver.py index 466f5dd9..1855eca1 100644 --- a/tests/test_fuzz_atheris_driver.py +++ b/tests/test_fuzz_atheris_driver.py @@ -24,7 +24,7 @@ def consume(data: bytes) -> None: accepted.append(data) target = Target("cli", consume, ("turbohtml.__main__.main",), (UnicodeError,)) - runtime = mocker.MagicMock(spec=["Setup", "Fuzz", "instrument_func"]) + runtime = mocker.MagicMock(spec=["Setup", "Fuzz", "instrument_func", "instrument_all"]) runtime.instrument_func.side_effect = lambda callback: callback mocker.patch("fuzz.atheris_driver.import_module", autospec=True, return_value=runtime) mocker.patch("fuzz.atheris_driver.rejection_hook", autospec=True, return_value=partial(rejected.append, 1)) @@ -36,6 +36,7 @@ def native_loop() -> None: runtime.Fuzz.side_effect = native_loop fuzz([target], ["turbohtml.__main__"], "cli", ["driver", "-runs=2"]) + runtime.instrument_all.assert_called_once_with() assert (accepted, rejected, runtime.Setup.call_args.args[0], runtime.Setup.call_args.kwargs) == ( [b"valid"], [1], diff --git a/tests/test_fuzz_atheris_parser_targets.py b/tests/test_fuzz_atheris_parser_targets.py new file mode 100644 index 00000000..13f7a66c --- /dev/null +++ b/tests/test_fuzz_atheris_parser_targets.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from fuzz.atheris_parser_targets import ( + document_observation, + fragment_observation, + incremental_observation, + parser_targets, + token_observation, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + from fuzz.atheris_registry import Target + + +@pytest.mark.parametrize( + "observe", + [ + pytest.param(document_observation, id="document"), + pytest.param(incremental_observation, id="incremental"), + ], +) +def test_atheris_parser_document_output(observe: Callable[[bytes], str]) -> None: + assert observe("

水😀

".encode()) == "

水😀

" + + +def test_atheris_parser_fragment_output() -> None: + assert fragment_observation(b"

x&y

") == "

x&y

" + + +def test_atheris_parser_token_fields() -> None: + assert token_observation(b'

x

') == ( + ("START_TAG", "p", None, (("id", "a"),), False, 1, 0), + ("TEXT", None, "x", None, False, 1, 10), + ("END_TAG", "p", None, (), False, 1, 11), + ) + + +@pytest.mark.parametrize("target", parser_targets(), ids=lambda target: target.name) +def test_atheris_parser_callback_rejects_invalid_utf8(target: Target) -> None: + with pytest.raises(UnicodeDecodeError): + target.callback(b"\xff") + + +def test_atheris_parser_exports_have_unique_consumers() -> None: + assert tuple((target.name, target.exports) for target in parser_targets()) == ( + ("html-document", ("turbohtml.parse", "turbohtml.Document", "turbohtml.Node")), + ( + "html-fragment", + ( + "turbohtml.parse_fragment", + "turbohtml.Element", + "turbohtml.Html", + "turbohtml.Formatter", + "turbohtml.Indent", + "turbohtml.Minify", + ), + ), + ("html-incremental", ("turbohtml.IncrementalParser",)), + ("html-tokenizer", ("turbohtml.tokenize", "turbohtml.Tokenizer", "turbohtml.Token", "turbohtml.TokenType")), + ) diff --git a/tests/test_fuzz_atheris_reference_targets.py b/tests/test_fuzz_atheris_reference_targets.py new file mode 100644 index 00000000..c579abfc --- /dev/null +++ b/tests/test_fuzz_atheris_reference_targets.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Final + +import pytest +from fuzz.atheris_reference_targets import ( + cli_observation, + computed_observation, + conformance_observation, + cssom_observation, + reference_targets, + render_observation, + schema_observation, + selector_observation, + transform_observation, +) + +from turbohtml import HTMLParseError +from turbohtml.convert import ExpressionError + +if TYPE_CHECKING: + from fuzz.atheris_registry import Target + + +def test_atheris_reference_rendered_text() -> None: + assert render_observation(b"

one two

") == ( + "one **two**", + "one two", + "one two", + b"

one two

", + ) + + +def test_atheris_reference_annotation_text() -> None: + markdown, plain, tagged, canonical = render_observation(b'

one

') + assert (markdown, plain, tagged, canonical) == ( + "[one](/x)", + "one", + "one", + b'

one

', + ) + + +def test_atheris_reference_conformance_records() -> None: + valid, codes, diagnostic = conformance_observation( + b'x' + ) + assert (valid, "img-missing-alt" in codes, diagnostic) == (False, True, None) + + +def test_atheris_reference_css_records() -> None: + assert cssom_observation(b"color:red;color:blue!important;margin:0") == ( + ("color", "blue", True), + ("margin", "0", False), + ) + + +def test_atheris_reference_empty_css_records() -> None: + assert cssom_observation(b"") == () + + +def test_atheris_reference_selector_specificity() -> None: + expression, specificity = selector_observation(b"div#a > p.x") + assert (bool(expression), specificity) == (True, ((1, 1, 2),)) + + +def test_atheris_reference_untranslatable_selector() -> None: + with pytest.raises(ExpressionError): + selector_observation(b":dir(ltr)") + + +@pytest.mark.parametrize( + ("source", "expected"), + [ + pytest.param(b"one", (True, True), id="valid"), + pytest.param(b"", (False, False), id="invalid"), + ], +) +def test_atheris_reference_schema_verdict(source: bytes, expected: tuple[bool, bool]) -> None: + assert schema_observation(source) == expected + + +@pytest.mark.parametrize( + "source", + [pytest.param(b"one", id="plain"), pytest.param(b"\x00", id="null"), pytest.param(b"""a'b"c""", id="both-quotes")], +) +def test_atheris_reference_transform_parameter(source: bytes) -> None: + assert transform_observation(source) == source.decode() + + +def test_atheris_reference_cli_file_output() -> None: + assert cli_observation(b"

one
two

") == "one\ntwo" + + +@pytest.mark.parametrize("target", reference_targets(), ids=lambda target: target.name) +def test_atheris_reference_callback_rejects_invalid_utf8(target: Target) -> None: + with pytest.raises(UnicodeDecodeError): + target.callback(b"\xff") + + +def test_atheris_reference_strict_parse_record() -> None: + assert conformance_observation(b" None: + assert computed_observation(source) == expected + + +def test_atheris_reference_xml_rejection_is_documented() -> None: + target: Final = next(target for target in reference_targets() if target.name == "xml-schema") + assert target.exceptions == (UnicodeDecodeError, HTMLParseError) + with pytest.raises(HTMLParseError): + target.callback(b"") diff --git a/tests/test_fuzz_atheris_targets.py b/tests/test_fuzz_atheris_targets.py new file mode 100644 index 00000000..6caa03fc --- /dev/null +++ b/tests/test_fuzz_atheris_targets.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Final + +import pytest +from fuzz.atheris_registry import validate_owners +from fuzz.atheris_targets import MODULES, main, owner_inventory, public_targets + +if TYPE_CHECKING: + from pathlib import Path + + from pytest_mock import MockerFixture + + +def test_atheris_targets_complete_qualified_inventory() -> None: + owners: Final = owner_inventory() + assert ( + len(owners), + len(MODULES), + {name: owners[name] for name in ("turbohtml.parse", "turbohtml.parse_xml", "turbohtml.__main__.main")}, + ) == ( + 209, + 20, + { + "turbohtml.parse": "html-document", + "turbohtml.parse_xml": "xml-schema", + "turbohtml.__main__.main": "html-cli", + }, + ) + + +def test_atheris_targets_missing_group_is_rejected() -> None: + with pytest.raises(ValueError, match="missing="): + validate_owners(public_targets()[:-1], MODULES) + + +def test_atheris_targets_duplicate_group_is_rejected() -> None: + targets: Final = public_targets() + with pytest.raises(ValueError, match="Duplicate export owners"): + validate_owners((*targets, targets[0]), MODULES) + + +def test_atheris_targets_cli_manifest_and_flags(tmp_path: Path, mocker: MockerFixture) -> None: + run: Final = mocker.patch("fuzz.atheris_targets.fuzz", autospec=True) + corpus: Final = tmp_path / "html-tokenizer" + assert main(("--target", "html-tokenizer", "--corpus", str(corpus), "-atheris_runs=64", "-max_len=128")) == 0 + assert (json.loads(corpus.with_suffix(".json").read_text()), run.call_args.args[2:]) == ( + { + "target": "html-tokenizer", + "exports": ["turbohtml.Token", "turbohtml.TokenType", "turbohtml.Tokenizer", "turbohtml.tokenize"], + "corpus": str(corpus), + }, + ("html-tokenizer", (run.call_args.args[3][0], str(corpus), "-atheris_runs=64", "-max_len=128")), + ) diff --git a/tools/fuzz/atheris_content_targets.py b/tools/fuzz/atheris_content_targets.py new file mode 100644 index 00000000..e1cd0e2c --- /dev/null +++ b/tools/fuzz/atheris_content_targets.py @@ -0,0 +1,459 @@ +"""Fixed envelopes give content consumers independent expected results per input.""" + +from __future__ import annotations + +import html +import json +import unicodedata +from functools import partial +from typing import TYPE_CHECKING, Final, cast + +import turbohtml +from turbohtml import clean, detect, extract, parse_fragment +from turbohtml.migration import bleach, markupsafe, stdlib + +from .atheris_registry import Target +from .idna_nfc_oracles import idna_nfc_check +from .round_trip_oracles import ( + clean_url_check, + css_semantics_check, + encoding_decode_check, + encoding_stream_check, + fixpoint_check, + idna_host_check, + normalize_url_check, +) + +if TYPE_CHECKING: + from collections.abc import Callable + +__all__ = [ + "content_targets", + "minifier_check", + "minifier_observation", + "sanitizer_check", + "sanitizer_observation", + "stdlib_observation", + "url_check", +] + + +def content_targets() -> tuple[Target, ...]: + """Keep each qualified content export attached to an executable contract.""" + return ( + Target( + "content-sanitize", + _sanitize, + tuple( + f"turbohtml.clean.{name}" + for name in ( + "DEFAULT_ATTRIBUTES", + "DEFAULT_CSS_PROPERTIES", + "DEFAULT_SCHEMES", + "DEFAULT_TAGS", + "OnDisallowed", + "Policy", + "Removed", + "Sanitizer", + "Transform", + "sanitize", + "sanitize_node", + "sanitize_report", + "sanitize_report_node", + "collapse_whitespace_node", + "strip_comments_node", + "transform_node", + ) + ), + ), + Target( + "content-links", + _links, + tuple( + f"turbohtml.clean.{name}" + for name in ( + "DEFAULT_CALLBACKS", + "DEFAULT_PHONE_LABELS", + "Callback", + "LinkCandidate", + "LinkDetector", + "LinkSpan", + "Linker", + "Linkify", + "PhoneFormat", + "PhoneGrouping", + "PhoneNumber", + "PhoneNumbers", + "PhoneType", + "linkify", + "linkify_node", + "nofollow", + "target_blank", + ) + ), + ), + Target( + "content-minify", + _minify, + tuple( + f"turbohtml.clean.{name}" + for name in ( + "CSSMinify", + "JSMinify", + "Minify", + "minify", + "minify_css", + "minify_css_inline", + "minify_js", + ) + ), + ), + Target( + "content-detect", + _detect, + tuple( + f"turbohtml.detect.{name}" + for name in ( + "Detection", + "EncodingDetector", + "EncodingMatch", + "LanguageDetection", + "LanguageMatch", + "NormalizationForm", + "detect", + "detect_all", + "detect_language", + "is_normalized", + "normalize", + ) + ), + ), + Target( + "content-extract", + _extract, + ( + *( + f"turbohtml.extract.{name}" + for name in ( + "Article", + "DateExtraction", + "Element", + "Entry", + "Extraction", + "Feed", + "Link", + "MicrodataItem", + "OpenGraph", + "Paragraph", + "PublicationDate", + "RdfaItem", + "StructuredData", + "UrlCleaning", + "boilerplate", + "clean_url", + "dates", + "extract_links", + "feed", + "microdata", + "normalize_url", + "opengraph", + ) + ), + *( + f"turbohtml.{name}" + for name in ( + "Article", + "Entry", + "Feed", + "Link", + "MicrodataItem", + "OpenGraph", + "RdfaItem", + "StructuredData", + ) + ), + ), + ), + Target( + "content-migration", + _migration, + ( + *( + f"turbohtml.migration.bleach.{name}" + for name in ("ALLOWED_ATTRIBUTES", "ALLOWED_PROTOCOLS", "ALLOWED_TAGS", "attribute_policy", "clean") + ), + *( + f"turbohtml.migration.markupsafe.{name}" + for name in ("EscapeFormatter", "Markup", "escape", "escape_silent", "soft_str") + ), + ), + ), + Target("content-stdlib", _stdlib, ("turbohtml.migration.stdlib.HTMLParser",)), + ) + + +def _sanitize(data: bytes) -> None: + sanitizer_check(data) + text: Final = _text(data) + markup: Final = f'{text}{text}' + options: Final = clean.Policy( + tags=clean.DEFAULT_TAGS, + attributes=clean.DEFAULT_ATTRIBUTES, + url_schemes=clean.DEFAULT_SCHEMES, + css_properties=clean.DEFAULT_CSS_PROPERTIES, + on_disallowed_tag=clean.OnDisallowed.STRIP, + transform_tags={"unknown": clean.Transform("b")}, + ) + expected: Final = f"{text}{text}" + _equal(sanitizer_observation(data), (expected, [clean.Removed("b", "title"), clean.Removed("b", "onclick")])) + _equal(clean.sanitize(markup, options), expected) + sanitizer: Final = clean.Sanitizer(options) + _equal(sanitizer.sanitize(markup), expected) + rendered, removed = clean.sanitize_report(markup, options) + _equal(rendered, expected) + _equal(removed, [clean.Removed("b", "title"), clean.Removed("b", "onclick")]) + _equal(sanitizer.sanitize_report(markup), (rendered, removed)) + _equal(clean.sanitize_node(parse_fragment(markup), options).inner_html, expected) + node, report = clean.sanitize_report_node(parse_fragment(markup), options) + _equal((node.inner_html, report), (expected, removed)) + _equal(sanitizer.sanitize_node(parse_fragment(markup)).inner_html, expected) + node, report = sanitizer.sanitize_report_node(parse_fragment(markup)) + _equal((node.inner_html, report), (expected, removed)) + node = parse_fragment(f"

{text} x

") + _equal(clean.transform_node(node, clean.strip_comments_node, clean.collapse_whitespace_node).text, f"{text} x") + + +def _links(data: bytes) -> None: + text: Final = _text(data) + url: Final = f"https://example.com/{text}" + candidate: Final = clean.LinkCandidate(url, text) + callback: Final[clean.Callback] = clean.target_blank + _equal(callback(clean.nofollow(candidate)).attrs, {"rel": "nofollow", "target": "_blank"}) + options: Final = clean.Linkify(callbacks=(*clean.DEFAULT_CALLBACKS, callback)) + linked: Final = f'{url}' + _equal(clean.linkify(url, options), linked) + _equal(clean.Linker(options).linkify(url), linked) + _equal(clean.linkify_node(parse_fragment(url), options).inner_html, linked) + _equal(clean.Linker(options).linkify_node(parse_fragment(url)).inner_html, linked) + _equal(clean.LinkDetector().find(url), [clean.LinkSpan(0, len(url), url, url, is_email=False)]) + phones: Final = clean.PhoneNumbers( + regions=("US",), + grouping=clean.PhoneGrouping.EXACT, + types=(clean.PhoneType.FIXED_LINE_OR_MOBILE,), + ignore_numbers_after=clean.DEFAULT_PHONE_LABELS, + ) + number: Final = clean.PhoneNumber.parse("650-253-0000", regions=("US",)) + _equal(number, clean.PhoneNumber(1, "6502530000", None, "US", clean.PhoneType.FIXED_LINE_OR_MOBILE)) + _equal(number.format(clean.PhoneFormat.E164), "+16502530000") + spans: Final = clean.LinkDetector(phones=phones).find(f"{text} 650-253-0000") + _equal([span.phone for span in spans], [number]) + + +def _minify(data: bytes) -> None: + minifier_check(data) + text: Final = _text(data) + _equal( + minifier_observation(data), + ( + f"

{text}

", + f".{text}{{color:red}}", + "color:red", + f'const value="{text}"', + ), + ) + + +def _detect(data: bytes) -> None: + text: Final = _text(data) + _equal(encoding_stream_check(f"utf-8-bom\n{text}"), None) + _equal(encoding_decode_check(f"utf-8-bom\n{text}"), None) + encoded: Final = b"\xef\xbb\xbf" + text.encode() + options: Final = detect.Detection() + match: Final = detect.detect(encoded, options) + _equal(match, detect.EncodingMatch("UTF-8-SIG", 1.0, None, bom=True, codec="whatwg-utf-8-sig")) + _equal(detect.detect_all(encoded, options)[0], match) + stream: Final = detect.EncodingDetector(options) + stream.feed(b"") + for chunk in encoded: + stream.feed(bytes((chunk,))) + _equal(stream.close(), match) + language: Final = detect.detect_language(str(len(data)), detect.LanguageDetection(allowed=frozenset({"eng"}))) + _equal(language, detect.LanguageMatch(None, 0.0, None)) + for form in cast("tuple[detect.NormalizationForm, ...]", ("NFC", "NFD", "NFKC", "NFKD")): + normalized: Final = detect.normalize(form, f"{text}e\u0301") + _equal(normalized, unicodedata.normalize(form, f"{text}e\u0301")) + _equal(detect.is_normalized(form, normalized), expected=True) + + +def _extract(data: bytes) -> None: + url_check(data) + text: Final = _text(data) + url: Final = f"https://example.com/{text}" + markup: Final = ( + f'{text}' + f'' + f'' + f'

{text}

{text}' + f'
{text}
' + f'
{text}
' + ) + document: Final = turbohtml.parse(markup) + article: Final = document.article() + _equal(isinstance(article, (extract.Article, turbohtml.Article)), expected=True) + _equal(article.title, text) + links: Final = document.links() + _equal(isinstance(links[0], (extract.Link, turbohtml.Link)), expected=True) + _equal(links[0].url, url) + _equal(extract.extract_links(markup), {url}) + _equal(extract.normalize_url(url, extract.UrlCleaning()), url) + _equal(extract.clean_url(url, extract.UrlCleaning()), url) + paragraphs: Final = extract.boilerplate(f"

{text}

", extract.Extraction(min_length=0)) + _equal(isinstance(paragraphs[0], extract.Paragraph), expected=True) + _equal(paragraphs[0].text, text) + _equal(isinstance(document.select_one("article"), extract.Element), expected=True) + items: Final = extract.microdata(markup) + _equal(isinstance(items[0], (extract.MicrodataItem, turbohtml.MicrodataItem)), expected=True) + _equal(items[0].get_all("name"), [text]) + _equal(json.loads(items[0].json())["properties"]["name"], [text]) + graph: Final = extract.opengraph(markup) + _equal(isinstance(graph, (extract.OpenGraph, turbohtml.OpenGraph)), expected=True) + _equal((graph["title"], graph.is_valid()), (text, True)) + structured: Final = document.structured_data() + _equal(isinstance(structured, (extract.StructuredData, turbohtml.StructuredData)), expected=True) + _equal(isinstance(structured.rdfa[0], (extract.RdfaItem, turbohtml.RdfaItem)), expected=True) + _equal(structured.rdfa[0].get("https://schema.org/name"), text) + publication: Final = cast("extract.PublicationDate", extract.dates(markup, extract.DateExtraction())) + _equal(isinstance(publication, extract.PublicationDate), expected=True) + _equal(str(publication.date), "2020-01-02") + feed: Final = cast( + "extract.Feed", + extract.feed(f"{text}{text}"), + ) + _equal(isinstance(feed, (extract.Feed, turbohtml.Feed)), expected=True) + _equal((feed.title, feed.entries[0].title), (text, text)) + _equal(isinstance(feed.entries[0], (extract.Entry, turbohtml.Entry)), expected=True) + + +def _migration(data: bytes) -> None: + text: Final = _text(data) + _equal( + bleach.clean( + f"{text}", + tags=bleach.ALLOWED_TAGS, + attributes=bleach.ALLOWED_ATTRIBUTES, + protocols=bleach.ALLOWED_PROTOCOLS, + ), + f"{text}", + ) + names, predicate = bleach.attribute_policy({"b": ("title",)}) + _equal((names, predicate), ({"b": frozenset({"title"})}, None)) + raw: Final = f"<{text}>&\"'" + expected: Final = html.escape(raw).replace("'", "'").replace(""", """) + escaped: Final = markupsafe.escape(raw) + _equal((isinstance(escaped, markupsafe.Markup), str(escaped)), (True, expected)) + _equal(str(markupsafe.escape_silent(raw)), expected) + _equal(str(markupsafe.escape_silent(None)), "") + _equal(markupsafe.soft_str(escaped), escaped) + _equal(markupsafe.Markup(f"{text}").striptags(), text) + _equal(markupsafe.EscapeFormatter(markupsafe.escape).vformat("{}", (raw,), {}), expected) + + +def _stdlib(data: bytes) -> None: + text: Final = _text(data) + parser: Final = _TextParser() + markup: Final = f"{text}&" + parser.feed("") + for char in markup: + parser.feed(char) + parser.close() + _equal("".join(parser.text), f"{text}&") + _equal(stdlib_observation(data), f"{text}&") + parser.reset() + _equal(parser.getpos(), (1, 0)) + + +def minifier_check( + data: bytes, + css: Callable[[str], str] = clean.minify_css, + js: Callable[[str], str] | None = None, +) -> None: + """Stable minifier output must also preserve the stylesheet meaning.""" + text: Final = _text(data) + sheet: Final = f".{text} {{ color: #ff0000; margin: 0px 0px }}" + _equal(fixpoint_check(sheet, css, numeric=True), None) + _equal(css_semantics_check(f'

x

', css), None) + _equal( + fixpoint_check( + f'const value = "{text}";', + js or partial(clean.minify_js, options=clean.JSMinify(fold=False, mangle=False)), + numeric=False, + ), + None, + ) + + +def url_check(data: bytes, normalize: Callable[[str], str] = extract.normalize_url) -> None: + """Known-valid Unicode families distinguish stable wrong hosts from valid normalization.""" + affix: Final = data[:4].hex() + _equal(idna_host_check(f"acute-decomposed\n{affix}", normalize), None) + _equal(idna_nfc_check(f"reorder:{affix}", normalize), None) + url: Final = f"https://example.com/{_text(data)}" + _equal(normalize_url_check(url, normalize), None) + _equal(clean_url_check(url), None) + + +def sanitizer_observation(data: bytes) -> tuple[str, list[clean.Removed]]: + """Distinguish stripping from a no-op sanitizer with an explicit envelope.""" + text: Final = _text(data) + return clean.sanitize_report( + f'{text}{text}', + clean.Policy(on_disallowed_tag=clean.OnDisallowed.STRIP, transform_tags={"unknown": clean.Transform("b")}), + ) + + +def sanitizer_check( + data: bytes, + render: Callable[[bytes], tuple[str, list[clean.Removed]]] = sanitizer_observation, +) -> None: + """Reject wrong renderer results before treating an input as a valid oracle case.""" + text: Final = _text(data) + _equal(render(data), (f"{text}{text}", [clean.Removed("b", "title"), clean.Removed("b", "onclick")])) + + +def minifier_observation(data: bytes) -> tuple[str, str, str, str]: + """Keep printer results available for independent behavioral controls.""" + text: Final = _text(data) + return ( + clean.minify(f"

{text}

", clean.Minify(omit_optional_tags=False)), + clean.minify_css(f".{text} {{ color: red; }}", clean.CSSMinify()), + clean.minify_css_inline("color: red;", clean.CSSMinify()), + clean.minify_js(f'const value = "{text}";', clean.JSMinify()), + ) + + +def stdlib_observation(data: bytes) -> str: + """Chunked callbacks must retain decoded text rather than just complete parsing.""" + parser: Final = _TextParser() + for char in f"{_text(data)}&": + parser.feed(char) + parser.close() + return "".join(parser.text) + + +def _text(data: bytes) -> str: + return "value" + data[:32].hex() + + +def _equal(actual: object, expected: object) -> None: + if actual != expected: + msg = f"Content API mismatch: {actual!r} != {expected!r}" + raise AssertionError(msg) + + +class _TextParser(stdlib.HTMLParser): + def __init__(self) -> None: + super().__init__() + self.text: Final[list[str]] = [] + + def handle_data(self, data: str) -> None: + self.text.append(data) diff --git a/tools/fuzz/atheris_dom_targets.py b/tools/fuzz/atheris_dom_targets.py new file mode 100644 index 00000000..550fc26c --- /dev/null +++ b/tools/fuzz/atheris_dom_targets.py @@ -0,0 +1,444 @@ +"""Ordinary DOM consumers keep qualified export ownership executable.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from functools import partial +from html import escape +from typing import TYPE_CHECKING, Final, NamedTuple, cast + +import turbohtml as dom +from turbohtml import build, mutations, query, rewrite, saxparse, traverse, treebuild + +from .atheris_registry import Target + +if TYPE_CHECKING: + from collections.abc import Callable + + +def dom_targets() -> tuple[Target, ...]: + """Keep each corpus attached to the public operations it reaches.""" + return tuple( + Target(name, partial(_check, domain=name), exports, (UnicodeDecodeError,)) for name, exports in _OWNERS.items() + ) + + +def _check(data: bytes, *, domain: str) -> None: + dom_observation(data, domain).verify() + + +def dom_observation(data: bytes, domain: str) -> DomObservation: + """Keep constructed trees small for large corpus entries.""" + return _OBSERVERS[domain]( + data.decode("utf-8")[:64].replace("\x00", "\ufffd").replace("\r\n", "\n").replace("\r", "\n") + ) + + +def _construction(text: str) -> DomObservation: + attrs: Final[build.Attributes] = {"id": "item"} + content: Final[build.Content] = text + maker: Final = build.ElementMaker() + first: Final = maker("p", attrs, content) + second: Final = dom.ElementMaker()("p", attrs, content) + fragment: Final = dom.DocumentFragment() + fragment.append(dom.Text(text)) + root: Final = build.document(body=(build.E.p(text), dom.E.p(text), first, second)) + doctype: Final = next(child for child in root.children if isinstance(child, dom.Doctype)) + foreign: Final = cast("dom.Element", dom.parse_fragment("").find("rect")) + return DomObservation( + repr(( + root.find_all("p")[0].text, + second.text, + fragment.text, + dom.Comment(text).data, + dom.CData(text).data, + dom.ProcessingInstruction("target", text).data, + doctype.name, + foreign.namespace, + dom.Namespace.MATHML.value, + dom.Namespace.HTML.value, + )), + repr((text, text, text, text, text, text, "html", dom.Namespace.SVG, "math", "html")), + ) + + +def _traversal(text: str) -> DomObservation: + root: Final = build.E.div(build.E.p(text), build.E.b("tail")) + walked: Final = traverse.TreeWalker(root, traverse.NodeFilter.SHOW_ELEMENT) + iterator: Final = dom.NodeIterator(root, dom.NodeFilter.SHOW_ELEMENT) + root_walker: Final = dom.TreeWalker(root, dom.NodeFilter.SHOW_TEXT) + module_iterator: Final = traverse.NodeIterator(root, traverse.NodeFilter.SHOW_TEXT) + names: Final = [] + while (node := walked.next_node()) is not None: + names.append(cast("dom.Element", node).tag) + iter_names: Final = [] + while (node := iterator.next_node()) is not None: + iter_names.append(cast("dom.Element", node).tag) + texts: Final = [] + while (node := root_walker.next_node()) is not None: + texts.append(cast("dom.Text", node).data) + module_texts: Final = [] + while (node := module_iterator.next_node()) is not None: + module_texts.append(cast("dom.Text", node).data) + return DomObservation( + repr((names, iter_names, texts, module_texts, [item.tag for item in root.find_all(axis=dom.Axis.CHILDREN)])), + repr(( + ["p", "b"], + ["div", "p", "b"], + [text, "tail"], + [text, "tail"], + ["p", "b"], + )), + ) + + +def _query(text: str) -> DomObservation: + root: Final = build.E.div({"id": "root"}, build.E.p({"id": "item", "class": "picked"}, text)) + child: Final = cast("dom.Element", root.find("p")) + matching: Final = query.Matching(namespaces={"h": "http://www.w3.org/1999/xhtml"}, flags=0) + matcher: Final = query.Matcher("p", matching) + compiled: Final = query.compile("p") + smart: Final = cast("list[dom.XPathString]", dom.XPath("//@id", smart_strings=True)(root)) + direct: Final = dom.XPathString(text, child, is_attribute=False, attrname=None) + observed: Final = ( + [item.tag for item in query.select("p", root)], + query.select_one("p", root) == child, + [item.tag for item in query.iselect("p", root)], + [item.tag for item in query.filter("p", root)], + query.match("p", child), + query.closest("div", child) == root, + matcher.match(child), + compiled.select_one(root) == child, + query.Query(root).find("p").text(), + query.css("p").select(root)[0].text, + query.escape_identifier("a b"), + query.DEBUG, + matching.flags, + smart[0].attrname, + smart[0].getparent().tag, + str(direct), + direct.getparent().tag, + ) + invalid = "accepted" + try: + query.compile("[") + except query.SelectorSyntaxError: + invalid = "rejected" + return DomObservation( + repr((observed, invalid)), + repr(( + (["p"], True, ["p"], ["p"], True, True, True, True, text, text, "a\\ b", 1, 0, "id", "div", text, "p"), + "rejected", + )), + ) + + +def _mutation(text: str) -> DomObservation: + root: Final = build.E.div() + first: Final = dom.MutationObserver() + second: Final = mutations.MutationObserver() + for observer in (first, second): + observer.observe(root, child_list=True) + root.append(dom.Text(text)) + records: Final = first.take_records() + other: Final = second.take_records() + observed: Final = ( + [(record.type, record.added_nodes[0].text) for record in records], + [(record.type, record.added_nodes[0].text) for record in other], + all(isinstance(record, (dom.MutationRecord, mutations.MutationRecord)) for record in records + other), + ) + first.disconnect() + second.disconnect() + return DomObservation(repr(observed), repr(([("childList", text)], [("childList", text)], True))) + + +def _range(text: str) -> DomObservation: + root: Final = build.E.div(build.E.p(text), build.E.b("tail")) + boundary: Final = dom.Range(root) + boundary.set_end(root, 1) + snapshot: Final = dom.StaticRange(root, 0, root, 1) + cloned: Final = boundary.clone_contents() + return DomObservation( + repr(( + cloned.html, + boundary.start_offset, + boundary.end_offset, + snapshot.start_offset, + snapshot.end_offset, + snapshot.collapsed, + cloned.text, + )), + repr((f"

{escape(text, quote=False)}

", 0, 1, 0, 1, False, text)), + ) + + +def _shadow(text: str) -> DomObservation: + host: Final = build.E.div(dom.Text(text)) + shadow: Final = host.attach_shadow("open") + shadow.append(build.E.slot()) + return DomObservation( + repr(( + isinstance(shadow, dom.ShadowRoot), + shadow.mode, + shadow.host.tag, + shadow.html, + [node.text for node in cast("dom.Element", shadow.children[0]).assigned_nodes()], + )), + repr((True, "open", "div", "", [text])), + ) + + +def _locations(text: str) -> DomObservation: + source: Final = f'

{escape(text, quote=False)}

' + root: Final = dom.parse_fragment(source, source_locations=True) + child: Final = cast("dom.Element", root.find("p")) + location: Final = cast("dom.SourceLocation", child.source_location) + span: Final = location.start_tag + end: Final = cast("dom.SourceSpan", location.end_tag) + return DomObservation( + repr(( + isinstance(location, dom.SourceLocation), + isinstance(span, dom.SourceSpan), + source[span.start_offset : span.end_offset], + source[end.start_offset : end.end_offset], + source[location.attrs["id"].start_offset : location.attrs["id"].end_offset], + )), + repr((True, True, '

', "

", 'id="item"')), + ) + + +def _rewrite(text: str) -> DomObservation: + element_handler: Final[rewrite.ElementHandler] = _rewrite_element + text_handler: Final[rewrite.TextHandler] = partial(_rewrite_text, value=text) + comment_handler: Final[rewrite.CommentHandler] = _rewrite_comment + doctype_handler: Final[rewrite.DoctypeHandler] = _rewrite_doctype + return DomObservation( + rewrite.rewrite( + "

old

", + elements=(("p", element_handler),), + text=text_handler, + comments=comment_handler, + doctype=doctype_handler, + ), + '<!--before-->

' + escape(text, quote=False) + "

", + ) + + +def _rewrite_element(element: rewrite.Element) -> None: + element.set_attribute("id", "item") + + +def _rewrite_text(text: rewrite.Element, *, value: str) -> None: + text.set_text(value) + + +def _rewrite_comment(comment: rewrite.Element) -> None: + comment.set_text("new") + + +def _rewrite_doctype(doctype: rewrite.Element) -> None: + doctype.before("") + + +def _sax(text: str) -> DomObservation: + source: Final = f"

{escape(text, quote=False)}

" + events: Final[tuple[saxparse.SaxEvent, ...]] = tuple(saxparse.iter_events(source)) + collector: Final = _SaxCollector() + saxparse.sax_parse(source, collector) + expected: Final[tuple[saxparse.SaxEvent, ...]] = ( + saxparse.Doctype("html", None, None), + saxparse.ProcessingInstruction("target", "data"), + saxparse.Comment("note"), + saxparse.StartElement("html", ()), + saxparse.StartElement("head", ()), + saxparse.EndElement("head"), + saxparse.StartElement("body", ()), + saxparse.StartElement("p", ()), + *((saxparse.Characters(text),) if text else ()), + saxparse.EndElement("p"), + saxparse.EndElement("body"), + saxparse.EndElement("html"), + ) + return DomObservation(repr((events, tuple(collector.events))), repr((expected, expected))) + + +class _SaxCollector(saxparse.SaxHandler): + def __init__(self) -> None: + self.events: Final[list[saxparse.SaxEvent]] = [] + + def start_element(self, tag: str, attrs: tuple[tuple[str, str | None], ...]) -> None: + self.events.append(saxparse.StartElement(tag, attrs)) + + def end_element(self, tag: str) -> None: + self.events.append(saxparse.EndElement(tag)) + + def characters(self, data: str) -> None: + self.events.append(saxparse.Characters(data)) + + def comment(self, data: str) -> None: + self.events.append(saxparse.Comment(data)) + + def doctype(self, name: str, public_id: str | None, system_id: str | None) -> None: + self.events.append(saxparse.Doctype(name, public_id, system_id)) + + def processing_instruction(self, target: str, data: str) -> None: + self.events.append(saxparse.ProcessingInstruction(target, data)) + + +def _tree(text: str) -> DomObservation: + sink: Final[treebuild.TreeBuilder[_Built]] = _TreeText() + source: Final = f"

{escape(text, quote=False)}

" + + def element(tag: str) -> str: + return repr((tag, "http://www.w3.org/1999/xhtml", ())) + + expected: Final = ( + "document", + repr(("html", None, None)), + "pi:target:data", + "comment:note", + element("html"), + element("head"), + element("body"), + element("p"), + *(("text:" + text,) if text else ()), + ) + return DomObservation(repr(_flatten(treebuild.parse_into(source, sink))), repr(expected)) + + +def _flatten(node: _Built) -> tuple[str, ...]: + return (node.payload, *(value for child in node.children for value in _flatten(child))) + + +class _TreeText: + @staticmethod + def create_document() -> _Built: + return _Built("document") + + @staticmethod + def create_doctype(name: str, public_id: str | None, system_id: str | None) -> _Built: + return _Built(repr((name, public_id, system_id))) + + @staticmethod + def create_element(name: str, namespace: str, attrs: tuple[tuple[str, str | None], ...]) -> _Built: + return _Built(repr((name, namespace, attrs))) + + @staticmethod + def create_text(data: str) -> _Built: + return _Built("text:" + data) + + @staticmethod + def create_comment(data: str) -> _Built: + return _Built("comment:" + data) + + @staticmethod + def create_pi(target: str, data: str) -> _Built: + return _Built(f"pi:{target}:{data}") + + @staticmethod + def append(parent: _Built, child: _Built) -> None: + parent.children.append(child) + + +@dataclass +class _Built: + payload: str + children: list[_Built] = field(default_factory=list) + + +class DomObservation(NamedTuple): + """Expose both sides so corpus consumers can report a failed contract.""" + + actual: str + expected: str + + def verify(self) -> None: + """Keep rejected oracle results visible to the fuzz driver.""" + if self.actual != self.expected: + raise AssertionError((self.actual, self.expected)) + + +_OBSERVERS: Final[dict[str, Callable[[str], DomObservation]]] = { + "dom-construction": _construction, + "dom-traversal": _traversal, + "dom-query": _query, + "dom-mutation": _mutation, + "dom-range": _range, + "dom-shadow": _shadow, + "dom-locations": _locations, + "dom-rewrite": _rewrite, + "dom-sax": _sax, + "dom-treebuild": _tree, +} +_OWNERS: Final[dict[str, tuple[str, ...]]] = { + "dom-construction": tuple( + "turbohtml." + name + for name in ( + "CData", + "Comment", + "Doctype", + "DocumentFragment", + "E", + "ElementMaker", + "Namespace", + "ProcessingInstruction", + "Text", + ) + ) + + tuple("turbohtml.build." + name for name in ("Attributes", "Content", "E", "ElementMaker", "document")), + "dom-traversal": tuple("turbohtml." + name for name in ("Axis", "NodeFilter", "NodeIterator", "TreeWalker")) + + tuple("turbohtml.traverse." + name for name in ("NodeFilter", "NodeIterator", "TreeWalker")), + "dom-query": ( + "turbohtml.XPath", + "turbohtml.XPathString", + *tuple( + "turbohtml.query." + name + for name in ( + "DEBUG", + "Matcher", + "Matching", + "Query", + "SelectorSyntaxError", + "closest", + "compile", + "css", + "escape_identifier", + "filter", + "iselect", + "match", + "select", + "select_one", + ) + ), + ), + "dom-mutation": ( + "turbohtml.MutationObserver", + "turbohtml.MutationRecord", + "turbohtml.mutations.MutationObserver", + "turbohtml.mutations.MutationRecord", + ), + "dom-range": ("turbohtml.Range", "turbohtml.StaticRange"), + "dom-shadow": ("turbohtml.ShadowRoot",), + "dom-locations": ("turbohtml.SourceLocation", "turbohtml.SourceSpan"), + "dom-rewrite": tuple( + "turbohtml.rewrite." + name + for name in ("CommentHandler", "DoctypeHandler", "Element", "ElementHandler", "TextHandler", "rewrite") + ), + "dom-sax": tuple( + "turbohtml.saxparse." + name + for name in ( + "Characters", + "Comment", + "Doctype", + "EndElement", + "ProcessingInstruction", + "SaxEvent", + "SaxHandler", + "StartElement", + "iter_events", + "sax_parse", + ) + ), + "dom-treebuild": ("turbohtml.treebuild.TreeBuilder", "turbohtml.treebuild.parse_into"), +} +__all__ = ["DomObservation", "dom_observation", "dom_targets"] diff --git a/tools/fuzz/atheris_driver.py b/tools/fuzz/atheris_driver.py index 54dc7531..8110c92d 100644 --- a/tools/fuzz/atheris_driver.py +++ b/tools/fuzz/atheris_driver.py @@ -27,7 +27,8 @@ def fuzz( target: Final = {target.name: target for target in targets}[target_name] atheris: Final = import_module("atheris") instrument: Final = cast("Callable[[Callable[[bytes], None]], Callable[[bytes], None]]", atheris.instrument_func) - callback: Final = instrument(rejecting_callback(instrument(target.callback), target.exceptions, rejection_hook())) + callback: Final = instrument(rejecting_callback(target.callback, target.exceptions, rejection_hook())) + cast("Callable[[], None]", atheris.instrument_all)() cast("_Setup", atheris.Setup)(list(argv), callback, custom_mutator=custom_mutator) cast("Callable[[], None]", atheris.Fuzz)() diff --git a/tools/fuzz/atheris_parser_targets.py b/tools/fuzz/atheris_parser_targets.py new file mode 100644 index 00000000..9aa02933 --- /dev/null +++ b/tools/fuzz/atheris_parser_targets.py @@ -0,0 +1,177 @@ +"""Incremental public entry points must preserve the one-shot result.""" + +from __future__ import annotations + +from typing import Final, NamedTuple + +from turbohtml import ( + Document, + Element, + Formatter, + Html, + IncrementalParser, + Indent, + Minify, + Node, + Token, + Tokenizer, + TokenType, + parse, + parse_fragment, + tokenize, +) + +from .atheris_registry import Target + +__all__ = [ + "document_observation", + "fragment_observation", + "incremental_observation", + "parser_targets", + "token_observation", +] + + +def parser_targets() -> tuple[Target, ...]: + """Keep document and fragment grammar contexts distinct.""" + return ( + Target( + "html-document", + _document, + ("turbohtml.parse", "turbohtml.Document", "turbohtml.Node"), + (UnicodeDecodeError,), + ), + Target( + "html-fragment", + _fragment, + ( + "turbohtml.parse_fragment", + "turbohtml.Element", + "turbohtml.Html", + "turbohtml.Formatter", + "turbohtml.Indent", + "turbohtml.Minify", + ), + (UnicodeDecodeError,), + ), + Target("html-incremental", _incremental, ("turbohtml.IncrementalParser",), (UnicodeDecodeError,)), + Target( + "html-tokenizer", + _tokens, + ("turbohtml.tokenize", "turbohtml.Tokenizer", "turbohtml.Token", "turbohtml.TokenType"), + (UnicodeDecodeError,), + ), + ) + + +def _document(data: bytes) -> None: + document_observation(data) + + +def _fragment(data: bytes) -> None: + fragment_observation(data) + + +def _incremental(data: bytes) -> None: + incremental_observation(data) + + +def _tokens(data: bytes) -> None: + token_observation(data) + + +def document_observation(data: bytes) -> str: + """Exercise returned document records through serialization consumers.""" + document: Final = parse(data.decode("utf-8")) + _require("Document result type", condition=isinstance(document, (Document, Node))) + return _serialization(document) + + +def fragment_observation(data: bytes) -> str: + """Grammar fragments use a div context rather than document insertion modes.""" + fragment: Final = parse_fragment(data.decode("utf-8"), "div") + _require("Fragment result type", condition=isinstance(fragment, Element)) + return _serialization(fragment) + + +def incremental_observation(data: bytes) -> str: + """Empty feeds and multibyte boundaries must preserve the complete parse.""" + data.decode("utf-8") + parser: Final = IncrementalParser() + parser.feed(b"") + for value in data: + parser.feed(bytes((value,))) + parser.feed(b"") + document: Final = parser.close() + observed: Final = _serialization(document) + _require("Incremental document differs", condition=observed == parse(data).serialize()) + return observed + + +def token_observation(data: bytes) -> tuple[_TokenSnapshot, ...]: + """Streaming and reset must preserve token fields and source positions.""" + source: Final = data.decode("utf-8") + tokenizer: Final = Tokenizer() + streamed: Final = list(tokenizer.feed("")) + for char in source: + streamed.extend(tokenizer.feed(char)) + streamed.extend(tokenizer.feed("")) + streamed.extend(tokenizer.close()) + observed: Final = tuple(_token_snapshot(token) for token in streamed) + _require( + "Incremental tokens differ", condition=observed == tuple(_token_snapshot(token) for token in tokenize(source)) + ) + tokenizer.reset() + _require( + "Reset tokens differ", + condition=tuple(_token_snapshot(token) for token in tokenizer.feed(source)) + + tuple(_token_snapshot(token) for token in tokenizer.close()) + == observed, + ) + return observed + + +def _serialization(node: Node) -> str: + for formatter in Formatter: + for layout in (None, Indent(2)): + options: Final = Html(formatter=formatter, layout=layout) + rendered: Final = node.serialize(options) + _require("Iterator serialization differs", condition="".join(node.serialize_iter(options)) == rendered) + _require( + "Encoded serialization differs", condition=node.encode("utf-8", options) == rendered.encode("utf-8") + ) + minified_options: Final = Html(layout=Minify()) + _require( + "Minified encoding differs", + condition=node.encode("utf-8", minified_options) == node.serialize(minified_options).encode("utf-8"), + ) + return node.serialize() + + +def _token_snapshot(token: Token) -> _TokenSnapshot: + _require("Token result type", condition=isinstance(token, Token)) + _require("Token kind type", condition=isinstance(token.type, TokenType)) + return _TokenSnapshot( + token.type.name, + token.tag, + token.data, + tuple(token.attrs) if token.attrs is not None else None, + token.self_closing, + token.line, + token.col, + ) + + +def _require(message: str, *, condition: bool) -> None: + if not condition: + raise AssertionError(message) + + +class _TokenSnapshot(NamedTuple): + kind: str + tag: str | None + data: str | None + attrs: tuple[tuple[str, str], ...] | None + self_closing: bool + line: int + col: int diff --git a/tools/fuzz/atheris_reference_targets.py b/tools/fuzz/atheris_reference_targets.py new file mode 100644 index 00000000..f3acecbb --- /dev/null +++ b/tools/fuzz/atheris_reference_targets.py @@ -0,0 +1,333 @@ +"""Reference-shaped consumers must agree with their compiled public entry points.""" + +from __future__ import annotations + +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import Final, cast, get_args + +import turbohtml +from turbohtml import Canonical, HTMLParseError, Markdown, ParseError, PlainText, annotation_surface, annotation_tags +from turbohtml.__main__ import main +from turbohtml.conformance import ConformanceMessage, ConformanceReport, Severity, check, check_html +from turbohtml.convert import ( + ExpressionError, + GenericTranslator, + HTMLTranslator, + SelectorError, + css_specificity, + css_to_xpath, +) +from turbohtml.cssom import ComputedStyle, RuleList, StyleDeclaration, StyleRule, StyleSheet, computed_style +from turbohtml.transform import Transform, strparam, transform +from turbohtml.validate import RelaxNG, SchemaValidationError, ValidationError, ValidationResult, XMLSchema + +from .atheris_registry import Target + +__all__ = [ + "cli_observation", + "computed_observation", + "conformance_observation", + "cssom_observation", + "reference_targets", + "render_observation", + "schema_observation", + "selector_observation", + "transform_observation", +] + + +def reference_targets() -> tuple[Target, ...]: + """Keep each grammar and its documented rejection set independent.""" + return ( + Target( + "html-render", + _render, + ( + "turbohtml.Canonical", + "turbohtml.Markdown", + "turbohtml.PlainText", + "turbohtml.annotation_surface", + "turbohtml.annotation_tags", + "turbohtml.escape", + "turbohtml.unescape", + "turbohtml.__version__", + ), + (UnicodeDecodeError,), + ), + Target( + "html-conformance", + _conformance, + ( + "turbohtml.conformance.ConformanceMessage", + "turbohtml.conformance.ConformanceReport", + "turbohtml.conformance.Severity", + "turbohtml.conformance.check", + "turbohtml.conformance.check_html", + "turbohtml.HTMLParseError", + "turbohtml.ParseError", + ), + (UnicodeDecodeError,), + ), + Target( + "css-translate", + _selector, + ( + "turbohtml.convert.ExpressionError", + "turbohtml.convert.GenericTranslator", + "turbohtml.convert.HTMLTranslator", + "turbohtml.convert.SelectorError", + "turbohtml.convert.SelectorSyntaxError", + "turbohtml.convert.css_specificity", + "turbohtml.convert.css_to_xpath", + "turbohtml.SelectorSyntaxError", + ), + (UnicodeDecodeError, turbohtml.SelectorSyntaxError, ExpressionError), + ), + Target( + "css-object-model", + _cssom, + ( + "turbohtml.cssom.ComputedStyle", + "turbohtml.cssom.RuleList", + "turbohtml.cssom.StyleDeclaration", + "turbohtml.cssom.StyleRule", + "turbohtml.cssom.StyleSheet", + "turbohtml.cssom.computed_style", + ), + (UnicodeDecodeError,), + ), + Target( + "xml-schema", + _schema, + ( + "turbohtml.parse_xml", + "turbohtml.validate.RelaxNG", + "turbohtml.validate.SchemaValidationError", + "turbohtml.validate.ValidationError", + "turbohtml.validate.ValidationResult", + "turbohtml.validate.XMLSchema", + ), + (UnicodeDecodeError, HTMLParseError), + ), + Target( + "xml-transform", + _transform, + ( + "turbohtml.transform.Transform", + "turbohtml.transform.strparam", + "turbohtml.transform.transform", + ), + (UnicodeDecodeError,), + ), + Target("html-cli", _cli, ("turbohtml.__main__.main",), (UnicodeDecodeError,)), + ) + + +def _render(data: bytes) -> None: + render_observation(data) + + +def _conformance(data: bytes) -> None: + conformance_observation(data) + + +def _selector(data: bytes) -> None: + selector_observation(data) + + +def _cssom(data: bytes) -> None: + cssom_observation(data) + + +def _schema(data: bytes) -> None: + schema_observation(data) + + +def _transform(data: bytes) -> None: + transform_observation(data) + + +def _cli(data: bytes) -> None: + cli_observation(data) + + +def render_observation(data: bytes) -> tuple[str, str, str, bytes]: + """Escaping text must preserve it through a parsed element.""" + source: Final = data.decode("utf-8") + document: Final = turbohtml.parse(source) + _require("Escape round trip differs", condition=turbohtml.unescape(turbohtml.escape(source)) == source) + _require("Version result", condition=isinstance(turbohtml.__version__, str) and bool(turbohtml.__version__)) + plain: Final = document.to_text(PlainText()) + annotated, spans = document.to_annotated_text({"a": ("link",)}, PlainText()) + _require("Annotated text differs", condition=annotated == plain) + surface: Final = annotation_surface(annotated, spans) + _require( + "Annotation values", condition=all(isinstance(value, str) for values in surface.values() for value in values) + ) + return ( + document.to_markdown(Markdown()), + plain, + annotation_tags(annotated, spans), + document.canonicalize(Canonical()), + ) + + +def conformance_observation(data: bytes) -> tuple[bool, tuple[str, ...], tuple[str, int, int] | None]: + """Strict parse diagnostics and conformance reports use separate records.""" + source: Final = data.decode("utf-8") + report: Final = check(turbohtml.parse(source)) + _require("Conformance entry points differ", condition=report == check_html(source)) + _require("Conformance result type", condition=isinstance(report, ConformanceReport)) + _require("Conformance verdict differs", condition=bool(report) == (not report.errors)) + _require( + "Severity views differ", + condition=sorted(report.errors + report.warnings + report.infos) == sorted(report.messages), + ) + _require( + "Conformance record fields", + condition=all( + isinstance(message, ConformanceMessage) and message.severity in get_args(Severity) + for message in report.messages + ), + ) + try: + turbohtml.parse(source, strict=True) + except HTMLParseError as error: + _require("Parse error record", condition=isinstance(error.error, ParseError)) + diagnostic = (error.error.code, error.error.line, error.error.col) + else: + diagnostic = None + return report.valid, tuple(message.code for message in report.messages), diagnostic + + +def selector_observation(data: bytes) -> tuple[str, tuple[tuple[int, int, int], ...]]: + """Both translator wrappers must retain selector specificity and node selection.""" + selector: Final = data.decode("utf-8") + try: + expression: Final = css_to_xpath(selector) + except ExpressionError as error: + _require("Expression error ancestry", condition=isinstance(error, SelectorError)) + raise + _require("Generic translator differs", condition=GenericTranslator().css_to_xpath(selector) == expression) + _require("HTML translator differs", condition=HTMLTranslator().css_to_xpath(selector) == expression) + document: Final = turbohtml.parse('

one

two

') + _require( + "Selector translation differs", + condition=tuple(node.serialize() for node in document.select(selector)) + == tuple(node.serialize() for node in cast("list[turbohtml.Node]", document.xpath(expression))), + ) + return expression, tuple(css_specificity(selector)) + + +def cssom_observation(data: bytes) -> tuple[tuple[str, str, bool], ...]: + """Preserve declaration records through rule attachment.""" + source: Final = data.decode("utf-8") + declaration: Final = StyleDeclaration.parse(source) + observed: Final = tuple((name, declaration[name], declaration.important(name)) for name in declaration) + _require("Property order differs", condition=declaration.properties() == tuple(name for name, _, _ in observed)) + _require("Declaration length differs", condition=len(declaration) == len(observed)) + _require( + "Declaration lookup differs", + condition=all(name in declaration and declaration.get(name) == value for name, value, _ in observed), + ) + rebuilt: Final = StyleDeclaration(observed) + _require( + "Declaration reconstruction differs", + condition=tuple((name, rebuilt[name], rebuilt.important(name)) for name in rebuilt) == observed, + ) + rule: Final = StyleRule("p", declaration) + rules: Final = RuleList((rule,)) + _require("Rule wrapper differs", condition=tuple(rules) == (rules[0],) and rules[0].style is declaration) + sheet: Final = StyleSheet("p { " + declaration.text + " }") + _require( + "Sheet declaration differs", condition=tuple(item.style.text for item in sheet.rules) == (declaration.text,) + ) + computed_observation(data) + return observed + + +def computed_observation(data: bytes) -> str: + """Attach the declarations to the element whose cascade consumes them.""" + declaration: Final = StyleDeclaration.parse(data.decode("utf-8")) + document: Final = turbohtml.parse("

x

") + document.select("style")[0].set_text("p { color: black; " + declaration.text + " }") + computed: Final = computed_style(document.select("p")[0]) + _require("Computed result type", condition=isinstance(computed, ComputedStyle)) + _require( + "Computed lookup differs", + condition=all(name in computed and computed.get(name) == computed[name] for name in computed), + ) + _require( + "Computed reconstruction differs", + condition=ComputedStyle(tuple((name, computed[name]) for name in computed)).properties() + == computed.properties(), + ) + _require("Computed color missing", condition=computed.get("color") == computed["color"]) + return computed["color"] + + +def schema_observation(data: bytes) -> tuple[bool, bool]: + """Compiled schema verdicts and assertion errors must describe the same instance.""" + document: Final = turbohtml.parse_xml(data.decode("utf-8")) + schemas: Final = ( + XMLSchema( + '' + '' + ), + RelaxNG(''), + ) + verdicts: list[bool] = [] + for schema in schemas: + result: Final = schema.validate(document) + _require("Validation result type", condition=isinstance(result, ValidationResult)) + _require("Validation verdict differs", condition=bool(result) == schema.is_valid(document)) + _require( + "Validation error records", + condition=all(isinstance(error, ValidationError) and bool(error.message) for error in result.errors), + ) + try: + schema.assert_valid(document) + except SchemaValidationError as error: + _require("Validation assertion differs", condition=not result.valid and error.errors == result.errors) + else: + _require("Validation assertion accepted invalid instance", condition=result.valid) + verdicts.append(result.valid) + return verdicts[0], verdicts[1] + + +def transform_observation(data: bytes) -> str: + """Parameters must remain string values even when they contain XPath quotes.""" + source: Final = data.decode("utf-8") + stylesheet: Final = turbohtml.parse_xml( + '' + '' + '' + ) + document: Final = turbohtml.parse_xml("") + quoted: Final = strparam(source) + observed: Final = Transform(stylesheet, allow_imports=False)(document, value=quoted) + _require("Transform parameter differs", condition=observed == source) + _require( + "Transform entry points differ", + condition=transform(stylesheet, document, allow_imports=False, value=quoted) == observed, + ) + return observed + + +def cli_observation(data: bytes) -> str: + """File-based CLI output must match the library conversion.""" + source: Final = data.decode("utf-8") + with TemporaryDirectory(prefix="turbohtml-atheris-") as directory: + input_file: Final = Path(directory) / "input.html" + output_file: Final = Path(directory) / "output.txt" + input_file.write_text(source, encoding="utf-8") + _require("CLI exit status", condition=main(("to-text", str(input_file), "-o", str(output_file))) == 0) + observed: Final = output_file.read_text(encoding="utf-8") + _require("CLI result differs", condition=observed == turbohtml.parse(source).to_text()) + return observed + + +def _require(message: str, *, condition: bool) -> None: + if not condition: + raise AssertionError(message) diff --git a/tools/fuzz/atheris_targets.py b/tools/fuzz/atheris_targets.py new file mode 100644 index 00000000..153b7336 --- /dev/null +++ b/tools/fuzz/atheris_targets.py @@ -0,0 +1,82 @@ +"""Missing qualified owners must stop the driver before it starts the native runtime.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import TYPE_CHECKING, Final + +from .atheris_content_targets import content_targets +from .atheris_dom_targets import dom_targets +from .atheris_driver import fuzz +from .atheris_parser_targets import parser_targets +from .atheris_reference_targets import reference_targets +from .atheris_registry import validate_owners + +if TYPE_CHECKING: + from collections.abc import Sequence + + from .atheris_registry import Target + +__all__ = ["MODULES", "main", "owner_inventory", "public_targets"] + +MODULES: Final = ( + "turbohtml", + "turbohtml.__main__", + "turbohtml.build", + "turbohtml.clean", + "turbohtml.conformance", + "turbohtml.convert", + "turbohtml.cssom", + "turbohtml.detect", + "turbohtml.extract", + "turbohtml.migration.bleach", + "turbohtml.migration.markupsafe", + "turbohtml.migration.stdlib", + "turbohtml.mutations", + "turbohtml.query", + "turbohtml.rewrite", + "turbohtml.saxparse", + "turbohtml.transform", + "turbohtml.traverse", + "turbohtml.treebuild", + "turbohtml.validate", +) + + +def main(argv: Sequence[str] | None = None) -> int: + """LibFuzzer receives its flags after the target and corpus arguments.""" + targets: Final = public_targets() + parser: Final = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--target", required=True, choices=tuple(target.name for target in targets)) + parser.add_argument("--corpus", required=True, type=Path) + parsed: Final = parser.parse_known_args(argv) + arguments: Final = parsed[0] + flags: Final = parsed[1] + arguments.corpus.mkdir(parents=True, exist_ok=True) + manifest: Final = { + "target": arguments.target, + "exports": sorted(export for export, owner in owner_inventory().items() if owner == arguments.target), + "corpus": str(arguments.corpus), + } + arguments.corpus.with_suffix(".json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") + fuzz(targets, MODULES, arguments.target, (sys.argv[0], str(arguments.corpus), *flags)) + return 0 + + +def public_targets() -> tuple[Target, ...]: + """Each group owns separate modules and qualified re-export aliases.""" + targets: Final = parser_targets() + reference_targets() + content_targets() + dom_targets() + validate_owners(targets, MODULES) + return targets + + +def owner_inventory() -> dict[str, str]: + """Read executable consumers rather than assigning ownership from mode names.""" + return {export: target.name for export, target in validate_owners(public_targets(), MODULES).items()} + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tox.toml b/tox.toml index ab8b87a4..2f27e5de 100644 --- a/tox.toml +++ b/tox.toml @@ -270,16 +270,41 @@ commands_pre = [] # no extension install; the engine compiles standalone with t commands = [ [ "python", "{tox_root}{/}tools{/}js_sanitize.py" ] ] [env.atheris] -description = "verify native Atheris rejection against its released Linux x86_64 runtime" +description = "verify public consumers and native rejection against released Linux Atheris" base_python = [ "3.14" ] package = "skip" -deps = [ "atheris==3.1", "pytest>=9.1.1" ] +deps = [ + "atheris==3.1", + "markdown-it-py>=4.2", + "meson>=1.12.1", + "meson-python>=0.22.1", + "ninja>=1.13.2", + "pytest>=9.1.1", + "typing-extensions>=4.16", +] dependency_groups = [] -set_env = { PYTHONPATH = "{tox_root}/tools", PYTHONHASHSEED = "0" } +set_env = { PYTHONPATH = "{tox_root}/tools", PYTHONHASHSEED = "0", CC = "clang", CXX = "clang++", CFLAGS = "-fsanitize=fuzzer-no-link --coverage" } platform = "linux" -commands_pre = [] +commands_pre = [ + [ + "uv", + "pip", + "install", + "--reinstall", + "--no-deps", + "--no-build-isolation", + "--editable", + "{tox_root}", + "--config-settings=build-dir={env_dir}{/}cbuild", + "--config-settings=setup-args=-Dbuildtype=debug", + "--config-settings=compile-args=-j2" + ], +] commands = [ [ + "env", + "CC=gcc", + "CXX=g++", "python", "-m", "pytest", @@ -288,6 +313,7 @@ commands = [ "--basetemp={env_tmp_dir}/native", ] ] +allowlist_externals = [ "env" ] [env.bench] description = "benchmark turbohtml against each competitor isolated in its own uv venv (not run in CI)"