Coverage for /pythoncovmergedfiles/medio/medio/src/turbohtml/tools/fuzz/structure_generators.py: 55%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Reserve sibling costs before expansion so a leaf fallback can complete its input."""
3from __future__ import annotations
5import argparse
6import hashlib
7import random
8from collections import Counter
9from dataclasses import dataclass
10from html import escape
11from pathlib import Path
12from typing import TYPE_CHECKING, Final, cast
14if TYPE_CHECKING:
15 from collections.abc import Callable, Iterable, Sequence
18def main(argv: Sequence[str] | None = None) -> int:
19 """Keep corpus files as materialized inputs rather than oracle mode strings."""
20 parser: Final = argparse.ArgumentParser()
21 parser.add_argument("--output", type=Path, required=True)
22 parser.add_argument("--count", type=int, default=64)
23 parser.add_argument("--budget", type=int, default=30)
24 parser.add_argument("--seed", type=int, default=0)
25 parser.add_argument("--sweep", action="store_true")
26 args: Final = parser.parse_args(argv)
27 if args.count < 1:
28 parser.error("count must be positive")
29 try:
30 cases = (
31 generation_sweep(_HTML_GRAMMAR, args.count, args.budget)
32 if args.sweep
33 else tuple(
34 generate(_HTML_GRAMMAR, random.Random(args.seed + index), args.budget) for index in range(args.count)
35 )
36 )
37 except (GrammarError, BudgetError, ProductionFloorError) as error:
38 parser.error(str(error))
39 write_corpus(cases, args.output)
40 return 0
43def compile_grammar(
44 productions: Sequence[Production],
45 root: str,
46 *,
47 identifiers: tuple[str, ...] = ("x", "y", "z"),
48 materialize: Callable[[Generated], Generated] | None = None,
49) -> Grammar:
50 """Reject incomplete grammars before consuming a finite input budget."""
51 if (
52 not 1 <= len(identifiers) <= 64
53 or len(set(identifiers)) != len(identifiers)
54 or any(not name.isascii() or not name.isidentifier() for name in identifiers)
55 ):
56 msg = "identifier pools require 1 to 64 distinct ASCII identifiers"
57 raise GrammarError(msg)
58 rules: Final[dict[str, list[Production]]] = {}
59 names: Final[set[str]] = set()
60 for production in productions:
61 if production.name in names or production.nodes < 0 or production.height < 0:
62 msg = "production names must be unique, node costs nonnegative, and heights nonnegative"
63 raise GrammarError(msg)
64 names.add(production.name)
65 rules.setdefault(production.symbol, []).append(production)
66 if any(isinstance(part, Reference) and part.depth < 0 for part in production.parts):
67 msg = "reference depths must be nonnegative"
68 raise GrammarError(msg)
69 if root not in rules:
70 msg = f"unknown root: {root}"
71 raise GrammarError(msg)
72 symbols: Final = set(rules)
73 references: Final = {
74 part.symbol for production in productions for part in production.parts if isinstance(part, Reference)
75 }
76 if unknown := references - symbols:
77 msg = f"unknown symbols: {sorted(unknown)}"
78 raise GrammarError(msg)
79 if unreachable := symbols - _reachable(root, rules):
80 msg = f"unreachable symbols: {sorted(unreachable)}"
81 raise GrammarError(msg)
82 minimum: Final[dict[str, tuple[int, int]]] = {}
83 while _relax(rules, minimum):
84 pass
85 if missing := symbols - minimum.keys():
86 msg = f"nonproductive symbols: {sorted(missing)}"
87 raise GrammarError(msg)
88 return Grammar(
89 tuple(productions),
90 root,
91 {name: tuple(items) for name, items in rules.items()},
92 minimum,
93 identifiers,
94 _edges(productions, minimum),
95 materialize,
96 )
99def _edges(
100 productions: Sequence[Production], minimum: dict[str, tuple[int, int]]
101) -> tuple[tuple[Production, int, str, int, int], ...]:
102 """Price every reference once as its production's minimum cost less the referenced child's minimum."""
103 edges: Final[list[tuple[Production, int, str, int, int]]] = []
104 for production in productions:
105 base: Final = _cost(production, minimum)
106 for index, part in enumerate(production.parts):
107 if isinstance(part, Reference):
108 child: Final = minimum[part.symbol]
109 edges.append((production, index, part.symbol, base[0] - child[0], base[1] - child[1]))
110 return tuple(edges)
113def _reachable(root: str, rules: dict[str, list[Production]]) -> set[str]:
114 found: Final[set[str]] = set()
115 pending: Final = [root]
116 while pending:
117 if (symbol := pending.pop()) not in found:
118 found.add(symbol)
119 pending.extend(
120 part.symbol for production in rules[symbol] for part in production.parts if isinstance(part, Reference)
121 )
122 return found
125def _relax(rules: dict[str, list[Production]], minimum: dict[str, tuple[int, int]]) -> bool:
126 changed = False
127 for symbol, productions in rules.items():
128 for production in productions:
129 references: Final = [part.symbol for part in production.parts if isinstance(part, Reference)]
130 if all(part in minimum for part in references):
131 cost: Final = _cost(production, minimum)
132 if symbol not in minimum or cost < minimum[symbol]:
133 minimum[symbol] = cost
134 changed = True
135 return changed
138def _cost(production: Production, minimum: dict[str, tuple[int, int]]) -> tuple[int, int]:
139 children: Final = [minimum[part.symbol] for part in production.parts if isinstance(part, Reference)]
140 return production.nodes + sum(cost[0] for cost in children), 1 + sum(cost[1] for cost in children)
143def _route(grammar: Grammar, name: str) -> tuple[tuple[Production, int], ...]:
144 if (target := next((production for production in grammar.productions if production.name == name), None)) is None:
145 msg = f"unknown production: {name}"
146 raise GrammarError(msg)
147 costs: Final = {target.symbol: _cost(target, grammar.minimum)}
148 paths: Final[dict[str, tuple[tuple[Production, int], ...]]] = {target.symbol: ((target, -1),)}
149 changed = True
150 while changed:
151 changed = False
152 for production, index, symbol, nodes, steps in grammar.edges:
153 if symbol in costs:
154 cost: Final = nodes + costs[symbol][0], steps + costs[symbol][1]
155 if production.symbol not in costs or cost < costs[production.symbol]:
156 costs[production.symbol] = cost
157 paths[production.symbol] = ((production, index), *paths[symbol])
158 changed = True
159 return paths[grammar.root]
162def _route_cost(grammar: Grammar, route: tuple[tuple[Production, int], ...]) -> tuple[int, int]:
163 cost = (0, 0)
164 for production, index in reversed(route):
165 base: Final = _cost(production, grammar.minimum)
166 if index < 0:
167 cost = base
168 else:
169 child: Final = grammar.minimum[cast("Reference", production.parts[index]).symbol]
170 cost = base[0] - child[0] + cost[0], base[1] - child[1] + cost[1]
171 return cost
174def generate(
175 grammar: Grammar,
176 rng: random.Random,
177 budget: int | GenerationBudget = 30,
178 *,
179 leaf: bool = False,
180 force: str | None = None,
181) -> Generated:
182 """Reserve the canonical minimum-node derivation and its expansion steps."""
183 route: Final = _route(grammar, force) if force is not None else ()
184 remaining = _initial_budget(grammar, budget, route)
185 pending: Final[list[tuple[bytes | Reference | Identifier, int, bool, tuple[tuple[Production, int], ...]]]] = [
186 (Reference(grammar.root, 0), 0, leaf, route)
187 ]
188 output: Final[list[bytes]] = []
189 fired: Final[list[str]] = []
190 bindings: Final = _Bindings({}, set(), set(), grammar.identifiers)
191 depth = 0
192 while pending:
193 part, level, minimal, path = pending.pop()
194 if isinstance(part, bytes):
195 output.append(part)
196 elif isinstance(part, Identifier):
197 output.append(_identifier(part, bindings))
198 else:
199 chosen, index, minimal = _choose(
200 grammar, rng, (part, level, minimal, path), _available(grammar, pending, remaining)
201 )
202 remaining = remaining[0] - chosen.nodes, remaining[1] - 1
203 if chosen.nodes:
204 depth = max(depth, level + chosen.height)
205 fired.append(chosen.name)
206 pending.extend(
207 (
208 item,
209 level + item.depth if isinstance(item, Reference) else level,
210 minimal,
211 path[1:] if path and position == index else (),
212 )
213 for position, item in reversed(tuple(enumerate(chosen.parts)))
214 )
215 if missing := bindings.references - bindings.definitions:
216 msg = f"undefined identifiers: {sorted(missing)}"
217 raise GrammarError(msg)
218 case: Final = Generated(
219 b"".join(output),
220 (budget.nodes if isinstance(budget, GenerationBudget) else budget) - remaining[0],
221 depth,
222 tuple(fired),
223 tuple(sorted(bindings.values.items())),
224 )
225 return case if grammar.materialize is None else grammar.materialize(case)
228def _initial_budget(
229 grammar: Grammar, budget: int | GenerationBudget, route: tuple[tuple[Production, int], ...]
230) -> tuple[int, int]:
231 remaining: Final = (
232 (budget.nodes, budget.steps)
233 if isinstance(budget, GenerationBudget)
234 else (budget, budget * max(cost[1] for cost in grammar.minimum.values()))
235 )
236 required: Final = _route_cost(grammar, route) if route else grammar.minimum[grammar.root]
237 if any(available < needed for available, needed in zip(remaining, required, strict=True)):
238 msg = "budget cannot complete the root"
239 raise BudgetError(msg)
240 return remaining
243def _available(
244 grammar: Grammar,
245 pending: list[tuple[bytes | Reference | Identifier, int, bool, tuple[tuple[Production, int], ...]]],
246 remaining: tuple[int, int],
247) -> tuple[int, int]:
248 reserved: Final = [
249 _route_cost(grammar, route) if route else grammar.minimum[item.symbol]
250 for item, _, _, route in pending
251 if isinstance(item, Reference)
252 ]
253 return remaining[0] - sum(cost[0] for cost in reserved), remaining[1] - sum(cost[1] for cost in reserved)
256def _identifier(part: Identifier, bindings: _Bindings) -> bytes:
257 if part.name not in bindings.values:
258 if len(bindings.values) == len(bindings.pool):
259 msg = "identifier pool exhausted"
260 raise GrammarError(msg)
261 bindings.values[part.name] = bindings.pool[len(bindings.values)]
262 if part.definition:
263 if part.name in bindings.definitions:
264 msg = f"duplicate identifier definition: {part.name}"
265 raise GrammarError(msg)
266 bindings.definitions.add(part.name)
267 else:
268 bindings.references.add(part.name)
269 return bindings.values[part.name].encode("ascii")
272def _choose(
273 grammar: Grammar,
274 rng: random.Random,
275 frame: tuple[Reference, int, bool, tuple[tuple[Production, int], ...]],
276 available: tuple[int, int],
277) -> tuple[Production, int, bool]:
278 part, level, minimal, path = frame
279 if path:
280 return path[0][0], path[0][1], True
281 eligible: Final = [
282 production
283 for production in grammar.rules[part.symbol]
284 if all(cost <= limit for cost, limit in zip(_cost(production, grammar.minimum), available, strict=True))
285 ]
286 if minimal or rng.random() >= 0.75 ** (level + 1):
287 return min(eligible, key=lambda production: _cost(production, grammar.minimum)), -1, True
288 return rng.choice(eligible), -1, False
291def generation_sweep(grammar: Grammar, minimum: int = 1, budget: int | GenerationBudget = 30) -> tuple[Generated, ...]:
292 """Force each declared production through a minimum-cost root path."""
293 if minimum < 1:
294 msg = "production floors must be positive"
295 raise ProductionFloorError(msg)
296 cases: Final = tuple(
297 generate(grammar, random.Random(index), budget, force=production.name)
298 for index, production in enumerate(production for production in grammar.productions for _ in range(minimum))
299 )
300 assert_production_floors(cases, (production.name for production in grammar.productions), minimum)
301 return cases
304def assert_production_floors(cases: Iterable[Generated], productions: Iterable[str], minimum: int = 1) -> Counter[str]:
305 """Fail a generation lane when a declared production stops executing."""
306 if minimum < 1:
307 msg = "production floors must be positive"
308 raise ProductionFloorError(msg)
309 counts: Final = Counter(name for case in cases for name in case.productions)
310 if missing := {name: counts[name] for name in productions if counts[name] < minimum}:
311 msg = f"production floors missed: {missing}"
312 raise ProductionFloorError(msg)
313 return counts
316def write_corpus(cases: Iterable[Generated], directory: Path) -> int:
317 """Hash materialized bytes so equal inputs share one corpus entry."""
318 directory.mkdir(parents=True, exist_ok=True)
319 paths: Final[set[Path]] = set()
320 for case in cases:
321 path: Final = directory / hashlib.sha256(case.data).hexdigest()
322 path.write_bytes(case.data)
323 paths.add(path)
324 return len(paths)
327@dataclass(frozen=True)
328class _Bindings:
329 values: dict[str, str]
330 definitions: set[str]
331 references: set[str]
332 pool: tuple[str, ...]
335@dataclass(frozen=True)
336class GenerationBudget:
337 """Bound lexical recursion separately from materialized DOM nodes."""
339 nodes: int
340 steps: int
343@dataclass(frozen=True)
344class Production:
345 """Charge the nodes emitted by fixed markup, including parser-created wrappers."""
347 name: str
348 symbol: str
349 parts: tuple[bytes | Reference | Identifier, ...]
350 nodes: int
351 origin: str
352 height: int = 0
355@dataclass(frozen=True)
356class Reference:
357 """Place child productions at their materialized tree depth."""
359 symbol: str
360 depth: int = 1
363@dataclass(frozen=True)
364class Identifier:
365 """Reuse allocated names so references bind to their generated definition."""
367 name: str
368 definition: bool
371@dataclass(frozen=True)
372class Grammar:
373 """Resolve minimum derivations before consuming an input's shared node budget."""
375 productions: tuple[Production, ...]
376 root: str
377 rules: dict[str, tuple[Production, ...]]
378 minimum: dict[str, tuple[int, int]]
379 identifiers: tuple[str, ...]
380 edges: tuple[tuple[Production, int, str, int, int], ...] # every reference, priced once for _route's relaxation
381 materialize: Callable[[Generated], Generated] | None = None
384@dataclass(frozen=True)
385class Generated:
386 """Keep source bytes beside their node costs and executed production trace."""
388 data: bytes
389 nodes: int
390 depth: int
391 productions: tuple[str, ...]
392 bindings: tuple[tuple[str, str], ...]
395class GrammarError(ValueError):
396 """Keep invalid grammar definitions outside the input corpus."""
399class BudgetError(ValueError):
400 """Reject budgets that cannot complete a minimum derivation."""
403class ProductionFloorError(ValueError):
404 """Treat unused productions as failed generation coverage."""
407def html_generate(rng: random.Random, budget: int = 30) -> Generated:
408 """Keep the general HTML oracle inside the shared node budget."""
409 return generate(_HTML_GRAMMAR, rng, budget)
412def html_grammar() -> Grammar:
413 """Separate child containers so adjacent text nodes cannot merge."""
414 productions: Final = [
415 Production(
416 "html:root",
417 "root",
418 (
419 b'<div id="',
420 Identifier("target", definition=True),
421 b'">',
422 Reference("node", 2),
423 b'<a href="#',
424 Identifier("target", definition=False),
425 b'">x</a></div>',
426 ),
427 4,
428 _HTML_STANDARD,
429 3,
430 ),
431 Production(
432 "html:nested",
433 "node",
434 (b"<div><span>", Reference("node", 2), b"</span><span>", Reference("node", 2), b"</span></div>"),
435 3,
436 _HTML_STANDARD,
437 1,
438 ),
439 ]
440 productions.extend(
441 Production(f"html:text:{index}", "node", (escape(text, quote=False).encode("utf-8"),), 1, _LOCAL_TABLES)
442 for index, text in enumerate(_HTML_TEXT)
443 )
444 for tag in _HTML_TAGS:
445 source, nodes, height = _element(tag)
446 productions.append(
447 Production(
448 f"html:tag:{tag}",
449 "node",
450 (source,),
451 nodes,
452 _DOMATO_TAGS if tag not in {"svg", "math", "script"} else _HTML_STANDARD,
453 height,
454 )
455 )
456 for name in _HTML_ATTRIBUTES:
457 for index, value in enumerate(_HTML_VALUES):
458 parts, nodes, height = _attribute(name, value, index)
459 productions.append(
460 Production(f"html:attribute:{name}:{index}", "node", parts, nodes, _LOCAL_TABLES, height)
461 )
462 return compile_grammar(productions, "root")
465def _element(tag: str) -> tuple[bytes, int, int]:
466 if source := _CONTEXTS.get(tag):
467 return source
468 if tag == "template":
469 return b"<template></template>", 2, 0
470 if tag in _VOID:
471 return f"<{tag}>".encode("ascii"), 1, 0
472 return f"<{tag}>x</{tag}>".encode("ascii"), 2, 1
475def _attribute(name: str, value: str, index: int) -> tuple[tuple[bytes | Reference | Identifier, ...], int, int]:
476 tag: Final = {"href": "a", "src": "img", "alt": "img", "name": "input", "type": "input", "value": "input"}.get(
477 name, "span"
478 )
479 encoded: Final = escape(
480 _IDENTIFIERS[1 + index % 2] if name == "id" else ("a", "b", "c")[index % 3] if name == "class" else value,
481 quote=True,
482 ).encode("utf-8")
483 parts: Final[tuple[bytes | Reference | Identifier, ...]] = (
484 (b"#", Identifier("target", definition=False)) if value == "#f" and name not in {"id", "class"} else (encoded,)
485 )
486 ending: Final = b">" if tag in _VOID else f">x</{tag}>".encode("ascii")
487 return (
488 (f'<{tag} {name}="'.encode("ascii"), *parts, b'"' + ending),
489 1 if tag in _VOID else 2,
490 0 if tag in _VOID else 1,
491 )
494_IDENTIFIERS: Final = ("x", "y", "z")
495_HTML_STANDARD: Final = "https://html.spec.whatwg.org/multipage/parsing.html"
496_DOMATO_TAGS: Final = (
497 "https://github.com/googleprojectzero/domato/blob/fadff396cc45d521cc594d3e2396e27e887b1963/html_tags.py"
498)
499_LOCAL_TABLES: Final = "https://github.com/tox-dev/turbohtml/blob/6a8722d7614a69df4b6ca6c8563f62d00f0837c1/tools/fuzz/round_trip_oracles.py"
500_CONTEXTS: Final = {
501 "table": (b"<table><tbody><tr><td>x</td></tr></tbody></table>", 5, 4),
502 "tr": (b"<table><tbody><tr><td>x</td></tr></tbody></table>", 5, 4),
503 "td": (b"<table><tbody><tr><td>x</td></tr></tbody></table>", 5, 4),
504 "th": (b"<table><tbody><tr><th>x</th></tr></tbody></table>", 5, 4),
505 "caption": (b"<table><caption>x</caption></table>", 3, 2),
506 "ul": (b"<ul><li>x</li></ul>", 3, 2),
507 "ol": (b"<ol><li>x</li></ol>", 3, 2),
508 "li": (b"<ul><li>x</li></ul>", 3, 2),
509 "dl": (b"<dl><dt>x</dt><dd>x</dd></dl>", 5, 2),
510 "dt": (b"<dl><dt>x</dt></dl>", 3, 2),
511 "dd": (b"<dl><dd>x</dd></dl>", 3, 2),
512 "select": (b"<select><option>x</option></select>", 3, 2),
513 "option": (b"<select><option>x</option></select>", 3, 2),
514}
516_HTML_TAGS: Final = (
517 "div", "p", "span", "a", "b", "i", "em", "strong", "code", "pre", "ul", "ol", "li", "table", "tr", "td", "th",
518 "h1", "h2", "blockquote", "section", "img", "br", "hr", "input", "label", "select", "option", "textarea", "svg",
519 "math", "template", "del", "sup", "script", "style", "title", "dl", "dt", "dd", "caption", "button", "form",
520) # fmt: skip
522_VOID: Final = frozenset({"img", "br", "hr", "input"})
524_HTML_TEXT: Final = (
525 "x", "hello world", "*", "_", "#", "|", "[", "]", "(", ")", "\\", "`", "&", "<", "<", ">", "1.", "-", "+",
526 " ", "\n", "\xa0", "é", "�", "😀", "\r\n", "~", "!", "=", "'", '"', " ", "---", "```", "> q",
527 "1) x", "a|b", "<!--c-->", "]]>", "--", " ", "\t", "**b**", "[l](u)", "<?pi x?>", "\x0c", ">", "\U0001f600",
528) # fmt: skip
530_HTML_ATTRIBUTES: Final = (
531 "id",
532 "class",
533 "href",
534 "src",
535 "title",
536 "lang",
537 "data-x",
538 "style",
539 "alt",
540 "name",
541 "type",
542 "value",
543)
545_HTML_VALUES: Final = ("a", "b c", "", "x y z", "&", '"', "'", "1", "#f", "javascript:x", "\n", "a b", "é", "<>")
548_HTML_GRAMMAR: Final = html_grammar()
550__all__ = [
551 "BudgetError",
552 "Generated",
553 "GenerationBudget",
554 "Grammar",
555 "GrammarError",
556 "Identifier",
557 "Production",
558 "ProductionFloorError",
559 "Reference",
560 "assert_production_floors",
561 "compile_grammar",
562 "generate",
563 "generation_sweep",
564 "html_generate",
565 "html_grammar",
566 "main",
567 "write_corpus",
568]
570if __name__ == "__main__":
571 raise SystemExit(main())