Coverage for /pythoncovmergedfiles/medio/medio/src/turbohtml/tools/fuzz/dom_targets.py: 97%

1"""Ordinary DOM consumers keep qualified export ownership executable.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from functools import partial 

7from html import escape 

8from typing import TYPE_CHECKING, Final, NamedTuple, cast 

9 

10import turbohtml as dom 

11from turbohtml import build, etree, mutations, query, rewrite, saxparse, traverse, treebuild 

12 

13from .atheris_registry import Target 

14 

15if TYPE_CHECKING: 

16 from collections.abc import Callable 

17 

18 

19def dom_targets() -> tuple[Target, ...]: 

20 """Keep each corpus attached to the public operations it reaches.""" 

21 return tuple( 

22 Target( 

23 name, 

24 partial(_check, domain=name), 

25 exports, 

26 (UnicodeDecodeError,), 

27 lossy_on_allocation_failure=name == "dom-mutation", 

28 ) 

29 for name, exports in _OWNERS.items() 

30 ) 

31 

32 

33def _check(data: bytes, *, domain: str) -> None: 

34 dom_observation(data, domain).verify() 

35 

36 

37def dom_observation(data: bytes, domain: str) -> DomObservation: 

38 """Keep constructed trees small for large corpus entries.""" 

39 return _OBSERVERS[domain]( 

40 data.decode("utf-8")[:64].replace("\x00", "\ufffd").replace("\r\n", "\n").replace("\r", "\n") 

41 ) 

42 

43 

44def _construction(text: str) -> DomObservation: 

45 attrs: Final[build.Attributes] = {"id": "item"} 

46 content: Final[build.Content] = text 

47 maker: Final = build.ElementMaker() 

48 first: Final = maker("p", attrs, content) 

49 second: Final = dom.ElementMaker()("p", attrs, content) 

50 fragment: Final = dom.DocumentFragment() 

51 fragment.append(dom.Text(text)) 

52 root: Final = build.document(body=(build.E.p(text), dom.E.p(text), first, second)) 

53 doctype: Final = next(child for child in root.children if isinstance(child, dom.Doctype)) 

54 foreign: Final = cast("dom.Element", dom.parse_fragment("<svg><rect></rect></svg>").find("rect")) 

55 return DomObservation( 

56 repr(( 

57 root.find_all("p")[0].text, 

58 second.text, 

59 fragment.text, 

60 dom.Comment(text).data, 

61 dom.CData(text).data, 

62 dom.ProcessingInstruction("target", text).data, 

63 doctype.name, 

64 foreign.namespace, 

65 dom.Namespace.MATHML.value, 

66 dom.Namespace.HTML.value, 

67 )), 

68 repr((text, text, text, text, text, text, "html", dom.Namespace.SVG, "math", "html")), 

69 ) 

70 

71 

72def _elementtree(text: str) -> DomObservation: 

73 root: Final = etree.Element("div", {"class": "one two"}) 

74 child: Final = etree.SubElement(root, "span") 

75 child.text = "prefix" + text 

76 child.tail = "tail" 

77 matches: Final = etree.XPath(".//span")(root) == [child] 

78 identity: Final = etree.ElementView(child.node) is child 

79 

80 @etree.document_context 

81 def detach() -> bool: 

82 root.remove(child) 

83 return child.getroottree().getroot() is root 

84 

85 context: Final = detach() 

86 root.append(child) 

87 etree.strip_tags(root, "span") 

88 etree.SubElement(root, "script").text = "discard" 

89 etree.strip_elements(root, "script") 

90 markup: Final = f"<p>prefix{escape(text)}</p>" 

91 document: Final = etree.document_fromstring(markup) 

92 snippet: Final = etree.fromstring(markup) 

93 fragment: Final = etree.fragment_fromstring(markup + "trailing") 

94 bridge: Final = etree.Element("p", {"data": text.encode().hex()}) 

95 copied: str | None = None 

96 html_attribute: str | None = None 

97 try: 

98 external = etree.to_lxml(bridge) 

99 except ModuleNotFoundError: 

100 # lxml is optional in the native fuzz runtime. 

101 pass 

102 else: 

103 copied = etree.tostring(etree.from_lxml(external), encoding="unicode") 

104 html_attribute = etree.to_lxml_html(bridge).get("data") 

105 return DomObservation( 

106 repr(( 

107 matches, 

108 identity, 

109 context, 

110 root.text, 

111 document.text_content(), 

112 snippet.text, 

113 fragment.tail, 

114 fragment.getparent(), 

115 copied, 

116 html_attribute, 

117 )), 

118 repr(( 

119 True, 

120 True, 

121 True, 

122 "prefix" + text + "tail", 

123 "prefix" + text, 

124 "prefix" + text, 

125 "trailing", 

126 None, 

127 f'<p data="{text.encode().hex()}"/>' if copied is not None else None, 

128 text.encode().hex() if copied is not None else None, 

129 )), 

130 ) 

131 

132 

133def _traversal(text: str) -> DomObservation: 

134 root: Final = build.E.div(build.E.p(text), build.E.b("tail")) 

135 walked: Final = traverse.TreeWalker(root, traverse.NodeFilter.SHOW_ELEMENT) 

136 iterator: Final = dom.NodeIterator(root, dom.NodeFilter.SHOW_ELEMENT) 

137 root_walker: Final = dom.TreeWalker(root, dom.NodeFilter.SHOW_TEXT) 

138 module_iterator: Final = traverse.NodeIterator(root, traverse.NodeFilter.SHOW_TEXT) 

139 names: Final = [] 

140 while (node := walked.next_node()) is not None: 

141 names.append(cast("dom.Element", node).tag) 

142 iter_names: Final = [] 

143 while (node := iterator.next_node()) is not None: 

144 iter_names.append(cast("dom.Element", node).tag) 

145 texts: Final = [] 

146 while (node := root_walker.next_node()) is not None: 

147 texts.append(cast("dom.Text", node).data) 

148 module_texts: Final = [] 

149 while (node := module_iterator.next_node()) is not None: 

150 module_texts.append(cast("dom.Text", node).data) 

151 return DomObservation( 

152 repr((names, iter_names, texts, module_texts, [item.tag for item in root.find_all(axis=dom.Axis.CHILDREN)])), 

153 repr(( 

154 ["p", "b"], 

155 ["div", "p", "b"], 

156 [text, "tail"], 

157 [text, "tail"], 

158 ["p", "b"], 

159 )), 

160 ) 

161 

162 

163def _query(text: str) -> DomObservation: 

164 root: Final = build.E.div({"id": "root"}, build.E.p({"id": "item", "class": "picked"}, text)) 

165 child: Final = cast("dom.Element", root.find("p")) 

166 matching: Final = query.Matching.soupsieve(namespaces={"h": "http://www.w3.org/1999/xhtml"}, flags=0) 

167 matcher: Final = query.Matcher("p", matching) 

168 compiled: Final = query.compile("p") 

169 smart: Final = cast("list[dom.XPathString]", dom.XPath("//@id", smart_strings=True)(root)) 

170 direct: Final = dom.XPathString(text, child, is_attribute=False, attrname=None) 

171 observed: Final = ( 

172 [item.tag for item in query.select("p", root)], 

173 query.select_one("p", root) == child, 

174 [item.tag for item in query.iselect("p", root)], 

175 [item.tag for item in query.filter("p", root)], 

176 query.match("p", child), 

177 query.closest("div", child) == root, 

178 matcher.match(child), 

179 compiled.select_one(root) == child, 

180 query.Query(root).find("p").text(), 

181 query.css("p").select(root)[0].text, 

182 query.escape_identifier("a b"), 

183 query.DEBUG, 

184 matching.flags, 

185 smart[0].attrname, 

186 smart[0].getparent().tag, 

187 str(direct), 

188 direct.getparent().tag, 

189 ) 

190 invalid = "accepted" 

191 try: 

192 query.compile("[") 

193 except query.SelectorSyntaxError: 

194 invalid = "rejected" 

195 return DomObservation( 

196 repr((observed, invalid)), 

197 repr(( 

198 (["p"], True, ["p"], ["p"], True, True, True, True, text, text, "a\\ b", 1, 0, "id", "div", text, "p"), 

199 "rejected", 

200 )), 

201 ) 

202 

203 

204def _mutation(text: str) -> DomObservation: 

205 root: Final = build.E.div() 

206 first: Final = dom.MutationObserver() 

207 second: Final = mutations.MutationObserver() 

208 for observer in (first, second): 

209 observer.observe(root, child_list=True) 

210 root.append(dom.Text(text)) 

211 records: Final = first.take_records() 

212 other: Final = second.take_records() 

213 observed: Final = ( 

214 [(record.type, record.added_nodes[0].text) for record in records], 

215 [(record.type, record.added_nodes[0].text) for record in other], 

216 all(isinstance(record, (dom.MutationRecord, mutations.MutationRecord)) for record in records + other), 

217 ) 

218 first.disconnect() 

219 second.disconnect() 

220 return DomObservation(repr(observed), repr(([("childList", text)], [("childList", text)], True))) 

221 

222 

223def _range(text: str) -> DomObservation: 

224 root: Final = build.E.div(build.E.p(text), build.E.b("tail")) 

225 boundary: Final = dom.Range(root) 

226 boundary.set_end(root, 1) 

227 snapshot: Final = dom.StaticRange(root, 0, root, 1) 

228 cloned: Final = boundary.clone_contents() 

229 escaped: Final = escape(text, quote=False).replace("\xa0", "&nbsp;") 

230 return DomObservation( 

231 repr(( 

232 cloned.html, 

233 boundary.start_offset, 

234 boundary.end_offset, 

235 snapshot.start_offset, 

236 snapshot.end_offset, 

237 snapshot.collapsed, 

238 cloned.text, 

239 )), 

240 repr((f"<p>{escaped}</p>", 0, 1, 0, 1, False, text)), 

241 ) 

242 

243 

244def _shadow(text: str) -> DomObservation: 

245 host: Final = build.E.div(dom.Text(text)) 

246 shadow: Final = host.attach_shadow("open") 

247 shadow.append(build.E.slot()) 

248 return DomObservation( 

249 repr(( 

250 isinstance(shadow, dom.ShadowRoot), 

251 shadow.mode, 

252 shadow.host.tag, 

253 shadow.html, 

254 [node.text for node in cast("dom.Element", shadow.children[0]).assigned_nodes()], 

255 )), 

256 repr((True, "open", "div", "<slot></slot>", [text])), 

257 ) 

258 

259 

260def _locations(text: str) -> DomObservation: 

261 source: Final = f'<p id="item">{escape(text, quote=False)}</p>' 

262 root: Final = dom.parse_fragment(source, source_locations=True) 

263 child: Final = cast("dom.Element", root.find("p")) 

264 location: Final = cast("dom.SourceLocation", child.source_location) 

265 span: Final = location.start_tag 

266 end: Final = cast("dom.SourceSpan", location.end_tag) 

267 return DomObservation( 

268 repr(( 

269 isinstance(location, dom.SourceLocation), 

270 isinstance(span, dom.SourceSpan), 

271 source[span.start_offset : span.end_offset], 

272 source[end.start_offset : end.end_offset], 

273 source[location.attrs["id"].start_offset : location.attrs["id"].end_offset], 

274 )), 

275 repr((True, True, '<p id="item">', "</p>", 'id="item"')), 

276 ) 

277 

278 

279def _rewrite(text: str) -> DomObservation: 

280 element_handler: Final[rewrite.ElementHandler] = _rewrite_element 

281 text_handler: Final[rewrite.TextHandler] = partial(_rewrite_text, value=text) 

282 comment_handler: Final[rewrite.CommentHandler] = _rewrite_comment 

283 doctype_handler: Final[rewrite.DoctypeHandler] = _rewrite_doctype 

284 return DomObservation( 

285 rewrite.rewrite( 

286 "<!doctype html><!--old--><p>old</p>", 

287 elements=(("p", element_handler),), 

288 text=text_handler, 

289 comments=comment_handler, 

290 doctype=doctype_handler, 

291 ), 

292 '&lt;!--before--&gt;<!doctype html><!--new--><p id="item">' + escape(text, quote=False) + "</p>", 

293 ) 

294 

295 

296def _rewrite_element(element: rewrite.Element) -> None: 

297 element.set_attribute("id", "item") 

298 

299 

300def _rewrite_text(text: rewrite.Element, *, value: str) -> None: 

301 text.set_text(value) 

302 

303 

304def _rewrite_comment(comment: rewrite.Element) -> None: 

305 comment.set_text("new") 

306 

307 

308def _rewrite_doctype(doctype: rewrite.Element) -> None: 

309 doctype.before("<!--before-->") 

310 

311 

312def _sax(text: str) -> DomObservation: 

313 source: Final = f"<!doctype html><?target data><!--note--><p>{escape(text, quote=False)}</p>" 

314 events: Final[tuple[saxparse.SaxEvent, ...]] = tuple(saxparse.iter_events(source)) 

315 collector: Final = _SaxCollector() 

316 saxparse.sax_parse(source, collector) 

317 expected: Final[tuple[saxparse.SaxEvent, ...]] = ( 

318 saxparse.Doctype("html", None, None), 

319 saxparse.ProcessingInstruction("target", "data"), 

320 saxparse.Comment("note"), 

321 saxparse.StartElement("html", ()), 

322 saxparse.StartElement("head", ()), 

323 saxparse.EndElement("head"), 

324 saxparse.StartElement("body", ()), 

325 saxparse.StartElement("p", ()), 

326 *((saxparse.Characters(text),) if text else ()), 

327 saxparse.EndElement("p"), 

328 saxparse.EndElement("body"), 

329 saxparse.EndElement("html"), 

330 ) 

331 observed: Final = (events, tuple(collector.events)) 

332 reference: Final = (expected, expected) 

333 if observed == reference: 

334 return DomObservation("matched", "matched") 

335 return DomObservation(repr(observed), repr(reference)) 

336 

337 

338class _SaxCollector(saxparse.SaxHandler): 

339 def __init__(self) -> None: 

340 self.events: Final[list[saxparse.SaxEvent]] = [] 

341 

342 def start_element(self, tag: str, attrs: tuple[tuple[str, str | None], ...]) -> None: 

343 self.events.append(saxparse.StartElement(tag, attrs)) 

344 

345 def end_element(self, tag: str) -> None: 

346 self.events.append(saxparse.EndElement(tag)) 

347 

348 def characters(self, data: str) -> None: 

349 self.events.append(saxparse.Characters(data)) 

350 

351 def comment(self, data: str) -> None: 

352 self.events.append(saxparse.Comment(data)) 

353 

354 def doctype(self, name: str, public_id: str | None, system_id: str | None) -> None: 

355 self.events.append(saxparse.Doctype(name, public_id, system_id)) 

356 

357 def processing_instruction(self, target: str, data: str) -> None: 

358 self.events.append(saxparse.ProcessingInstruction(target, data)) 

359 

360 

361def _tree(text: str) -> DomObservation: 

362 sink: Final[treebuild.TreeBuilder[_Built]] = _TreeText() 

363 source: Final = f"<!doctype html><?target data><!--note--><p>{escape(text, quote=False)}</p>" 

364 

365 def element(tag: str) -> str: 

366 return repr((tag, "http://www.w3.org/1999/xhtml", ())) 

367 

368 expected: Final = ( 

369 "document", 

370 repr(("html", None, None)), 

371 "pi:target:data", 

372 "comment:note", 

373 element("html"), 

374 element("head"), 

375 element("body"), 

376 element("p"), 

377 *(("text:" + text,) if text else ()), 

378 ) 

379 return DomObservation(repr(_flatten(treebuild.parse_into(source, sink))), repr(expected)) 

380 

381 

382def _flatten(node: _Built) -> tuple[str, ...]: 

383 return (node.payload, *(value for child in node.children for value in _flatten(child))) 

384 

385 

386class _TreeText: 

387 @staticmethod 

388 def create_document() -> _Built: 

389 return _Built("document") 

390 

391 @staticmethod 

392 def create_doctype(name: str, public_id: str | None, system_id: str | None) -> _Built: 

393 return _Built(repr((name, public_id, system_id))) 

394 

395 @staticmethod 

396 def create_element(name: str, namespace: str, attrs: tuple[tuple[str, str | None], ...]) -> _Built: 

397 return _Built(repr((name, namespace, attrs))) 

398 

399 @staticmethod 

400 def create_text(data: str) -> _Built: 

401 return _Built("text:" + data) 

402 

403 @staticmethod 

404 def create_comment(data: str) -> _Built: 

405 return _Built("comment:" + data) 

406 

407 @staticmethod 

408 def create_pi(target: str, data: str) -> _Built: 

409 return _Built(f"pi:{target}:{data}") 

410 

411 @staticmethod 

412 def append(parent: _Built, child: _Built) -> None: 

413 parent.children.append(child) 

414 

415 

416@dataclass 

417class _Built: 

418 payload: str 

419 children: list[_Built] = field(default_factory=list) 

420 

421 

422class DomObservation(NamedTuple): 

423 """Expose both sides so corpus consumers can report a failed contract.""" 

424 

425 actual: str 

426 expected: str 

427 

428 def verify(self) -> None: 

429 """Keep rejected oracle results visible to the fuzz driver.""" 

430 if self.actual != self.expected: 

431 raise AssertionError((self.actual, self.expected)) 

432 

433 

434_OBSERVERS: Final[dict[str, Callable[[str], DomObservation]]] = { 

435 "dom-construction": _construction, 

436 "dom-elementtree": _elementtree, 

437 "dom-traversal": _traversal, 

438 "dom-query": _query, 

439 "dom-mutation": _mutation, 

440 "dom-range": _range, 

441 "dom-shadow": _shadow, 

442 "dom-locations": _locations, 

443 "dom-rewrite": _rewrite, 

444 "dom-sax": _sax, 

445 "dom-treebuild": _tree, 

446} 

447_OWNERS: Final[dict[str, tuple[str, ...]]] = { 

448 "dom-construction": tuple( 

449 "turbohtml." + name 

450 for name in ( 

451 "CData", 

452 "Comment", 

453 "Doctype", 

454 "DocumentFragment", 

455 "E", 

456 "ElementMaker", 

457 "Namespace", 

458 "ProcessingInstruction", 

459 "Text", 

460 ) 

461 ) 

462 + tuple("turbohtml.build." + name for name in ("Attributes", "Content", "E", "ElementMaker", "document")), 

463 "dom-elementtree": tuple( 

464 "turbohtml.etree." + name 

465 for name in ( 

466 "Element", 

467 "ElementView", 

468 "SubElement", 

469 "XPath", 

470 "document_context", 

471 "document_fromstring", 

472 "fragment_fromstring", 

473 "from_lxml", 

474 "fromstring", 

475 "strip_elements", 

476 "strip_tags", 

477 "to_lxml", 

478 "to_lxml_html", 

479 "tostring", 

480 ) 

481 ), 

482 "dom-traversal": tuple("turbohtml." + name for name in ("Axis", "NodeFilter", "NodeIterator", "TreeWalker")) 

483 + tuple("turbohtml.traverse." + name for name in ("NodeFilter", "NodeIterator", "TreeWalker")), 

484 "dom-query": ( 

485 "turbohtml.XPath", 

486 "turbohtml.XPathString", 

487 *tuple( 

488 "turbohtml.query." + name 

489 for name in ( 

490 "DEBUG", 

491 "Matcher", 

492 "Matching", 

493 "Query", 

494 "SelectorSyntaxError", 

495 "closest", 

496 "compile", 

497 "css", 

498 "escape_identifier", 

499 "filter", 

500 "iselect", 

501 "match", 

502 "select", 

503 "select_one", 

504 ) 

505 ), 

506 ), 

507 "dom-mutation": ( 

508 "turbohtml.MutationObserver", 

509 "turbohtml.MutationRecord", 

510 "turbohtml.mutations.MutationObserver", 

511 "turbohtml.mutations.MutationRecord", 

512 ), 

513 "dom-range": ("turbohtml.Range", "turbohtml.StaticRange"), 

514 "dom-shadow": ("turbohtml.ShadowRoot",), 

515 "dom-locations": ("turbohtml.SourceLocation", "turbohtml.SourceSpan"), 

516 "dom-rewrite": tuple( 

517 "turbohtml.rewrite." + name 

518 for name in ("CommentHandler", "DoctypeHandler", "Element", "ElementHandler", "TextHandler", "rewrite") 

519 ), 

520 "dom-sax": tuple( 

521 "turbohtml.saxparse." + name 

522 for name in ( 

523 "Characters", 

524 "Comment", 

525 "Doctype", 

526 "EndElement", 

527 "ProcessingInstruction", 

528 "SaxEvent", 

529 "SaxHandler", 

530 "StartElement", 

531 "iter_events", 

532 "sax_parse", 

533 ) 

534 ), 

535 "dom-treebuild": ("turbohtml.treebuild.TreeBuilder", "turbohtml.treebuild.parse_into"), 

536} 

537__all__ = ["DomObservation", "dom_observation", "dom_targets"]