diff --git a/pyjelly/parse/decode.py b/pyjelly/parse/decode.py index 87b3af10..28c71cc3 100644 --- a/pyjelly/parse/decode.py +++ b/pyjelly/parse/decode.py @@ -309,6 +309,25 @@ def _adapter_missing(feature: str, *, stream_types: StreamTypes) -> Never: raise NotImplementedError(msg) +def _row_missing( + rows: str, + *, + allowed_in: tuple[jelly.PhysicalStreamType, ...], + stream_types: StreamTypes, +) -> Never: + """ + Raise the error for statement rows that the adapter does not decode. + + The adapter is chosen for the physical type of the stream, so the rows are + either not allowed in the stream, or the adapter is incomplete. + """ + if stream_types.physical_type not in allowed_in: + physical_type_name = jelly.PhysicalStreamType.Name(stream_types.physical_type) + msg = f"a stream of type {physical_type_name} cannot have {rows}" + raise JellyConformanceError(msg) + _adapter_missing(f"decoding {rows}", stream_types=stream_types) + + @mypyc_attr(allow_interpreted_subclasses=True) class Adapter(metaclass=ABCMeta): def __init__( @@ -350,19 +369,34 @@ def directional_literal(self, lex: str, language: str, direction: str) -> Any: # Optional abstract methods--not required to be implemented by all adapters def triple(self, terms: Iterable[Any]) -> Any: # noqa: ARG002 - _adapter_missing("decoding triples", stream_types=self.options.stream_types) + _row_missing( + "triples", + allowed_in=( + jelly.PHYSICAL_STREAM_TYPE_TRIPLES, + jelly.PHYSICAL_STREAM_TYPE_GRAPHS, + ), + stream_types=self.options.stream_types, + ) def quad(self, terms: Iterable[Any]) -> Any: # noqa: ARG002 - _adapter_missing("decoding quads", stream_types=self.options.stream_types) + _row_missing( + "quads", + allowed_in=(jelly.PHYSICAL_STREAM_TYPE_QUADS,), + stream_types=self.options.stream_types, + ) def graph_start(self, graph_id: Any) -> Any: # noqa: ARG002 - _adapter_missing( - "decoding graph start markers", stream_types=self.options.stream_types + _row_missing( + "graph start markers", + allowed_in=(jelly.PHYSICAL_STREAM_TYPE_GRAPHS,), + stream_types=self.options.stream_types, ) def graph_end(self) -> Any: - _adapter_missing( - "decoding graph end markers", stream_types=self.options.stream_types + _row_missing( + "graph end markers", + allowed_in=(jelly.PHYSICAL_STREAM_TYPE_GRAPHS,), + stream_types=self.options.stream_types, ) def namespace_declaration(self, name: str, iri: Any) -> Any: # noqa: ARG002 @@ -504,19 +538,27 @@ def decode_other_row(self, row_owner: jelly.RdfStreamRow, kind: str | None) -> N self.decode_graph_end(row_owner.graph_end) elif kind == "options": self.validate_stream_options(row_owner.options) + elif kind is None: + msg = "a row of the stream is empty" + raise JellyConformanceError(msg) else: msg = f"decoder not implemented for row {kind}" raise TypeError(msg) def validate_stream_options(self, options: jelly.RdfStreamOptions) -> None: + """Check that repeated stream options are the same as the first ones.""" stream_types, lookup_preset, params = self.options - assert stream_types.physical_type == options.physical_type - assert stream_types.logical_type == options.logical_type - assert params.stream_name == options.stream_name - assert params.version >= options.version - assert lookup_preset.max_prefixes == options.max_prefix_table_size - assert lookup_preset.max_datatypes == options.max_datatype_table_size - assert lookup_preset.max_names == options.max_name_table_size + if ( + stream_types.physical_type != options.physical_type + or stream_types.logical_type != options.logical_type + or params.stream_name != options.stream_name + or params.version < options.version + or lookup_preset.max_prefixes != options.max_prefix_table_size + or lookup_preset.max_datatypes != options.max_datatype_table_size + or lookup_preset.max_names != options.max_name_table_size + ): + msg = "the repeated stream options are not the same as the first ones" + raise JellyConformanceError(msg) def ingest_prefix_entry(self, entry: jelly.RdfPrefixEntry) -> None: """ @@ -644,7 +686,7 @@ def decode_literal(self, literal: jelly.RdfLiteral) -> Any: kind = _LITERAL_WHICH_ONEOF(literal, "literalKind") if kind == "langtag": language = literal.langtag or None - elif kind == "datatype" and self.datatypes.lookup_size: + elif kind == "datatype": datatype = self.datatypes.decode_datatype_term_index(literal.datatype) lex = literal.lex cache = self.literal_cache(language, datatype) @@ -705,7 +747,11 @@ def decode_namespace_declaration( return self.adapter.namespace_declaration(declaration.name, iri) def decode_graph_start(self, graph_start: jelly.RdfGraphStart) -> Any: - term = getattr(graph_start, graph_start.WhichOneof("graph")) + kind = graph_start.WhichOneof("graph") + if kind is None: + msg = "a graph start marker has no graph" + raise JellyConformanceError(msg) + term = getattr(graph_start, kind) return self.adapter.graph_start(self.decode_term(term)) def decode_graph_end(self, _: jelly.RdfGraphEnd) -> Any: @@ -722,7 +768,7 @@ def decode_statement(self, fields: list[Any], size: int) -> list[Any]: size (int): 3 for a triple, 4 for a quad Raises: - ValueError: if a missing repeated term is encountered + JellyConformanceError: if a missing repeated term is encountered Returns: Any: a list of decoded terms @@ -756,7 +802,7 @@ def decode_statement(self, fields: list[Any], size: int) -> list[Any]: for i in range(size): if terms[i] is _MISSING: msg = f"missing repeated term {_POSITIONS[i]}" - raise ValueError(msg) + raise JellyConformanceError(msg) return terms def decode_triple(self, triple: jelly.RdfTriple) -> Any: @@ -773,7 +819,7 @@ def decode_quoted_triple(self, triple: jelly.RdfTriple) -> Any: decoded_term = self.decode_term(jelly_term) else: msg = "repeated terms are not allowed in quoted triples" - raise ValueError(msg) + raise JellyConformanceError(msg) terms.append(decoded_term) return self.adapter.quoted_triple(terms) diff --git a/pyjelly/parse/ioutils.py b/pyjelly/parse/ioutils.py index c03e153d..331d37c8 100644 --- a/pyjelly/parse/ioutils.py +++ b/pyjelly/parse/ioutils.py @@ -2,12 +2,14 @@ import os from collections.abc import Generator, Iterator from itertools import chain -from typing import IO, Final, cast +from typing import IO, Final, TypeVar, cast from google.protobuf.descriptor import Descriptor, FieldDescriptor +from google.protobuf.message import DecodeError, Message from google.protobuf.proto import parse, parse_length_prefixed from pyjelly import jelly +from pyjelly.errors import JellyConformanceError from pyjelly.options import MIN_COLUMN_NAME_LOOKUP_SIZE, VERSION_1_2 from pyjelly.parse.decode import ParserOptions, options_from_frame, parser_options @@ -316,9 +318,49 @@ def detect_delimited( return inp, delimited_jelly_hint(header, tags) +# What protobuf raises for bytes it can't decode. The pure-Python implementation +# raises UnicodeDecodeError for a string that isn't UTF-8, the others DecodeError. +DECODE_ERRORS: Final = (DecodeError, UnicodeDecodeError) + +MessageT = TypeVar("MessageT", bound=Message) + + +def parse_delimited(message_class: type[MessageT], inp: IO[bytes]) -> MessageT | None: + """ + Read the next length-prefixed message, or None at the end of the input. + + Raises: + DecodeError: if the bytes are not a message of the class, or the input + ends in the middle of one. Or UnicodeDecodeError, see DECODE_ERRORS. + ValueError: if the input is closed. + + """ + try: + # protobuf annotates `input_bytes` as io.BytesIO, but any IO[bytes] works. + return parse_length_prefixed(message_class, inp) # type: ignore[arg-type] + except DECODE_ERRORS: + raise + except (ValueError, OverflowError) as error: + if getattr(inp, "closed", False): + raise + # The length prefix is cut short, or too large to read + msg = f"invalid length prefix ({error})" + raise DecodeError(msg) from None + + +def _not_jelly(error: Exception) -> JellyConformanceError: + msg = f"the stream is not valid Jelly-RDF ({error}). Is it cut short?" + return JellyConformanceError(msg) + + def frame_iterator(inp: IO[bytes]) -> Generator[jelly.RdfStreamFrame]: - # protobuf annotates `input_bytes` as io.BytesIO, but any IO[bytes] works. - while frame := parse_length_prefixed(jelly.RdfStreamFrame, inp): # type: ignore[arg-type] + while True: + try: + frame = parse_delimited(jelly.RdfStreamFrame, inp) + except DECODE_ERRORS as error: + raise _not_jelly(error) from None + if frame is None: + return yield frame @@ -371,7 +413,10 @@ def get_options_and_frames( options = options_from_frame(first_frame, delimited=True) return options, chain(skipped_frames, (first_frame,), frames) - frame = parse(jelly.RdfStreamFrame, inp.read()) + try: + frame = parse(jelly.RdfStreamFrame, inp.read()) + except DECODE_ERRORS as error: + raise _not_jelly(error) from None if not frame.rows: return _empty_stream_options(delimited=False), iter((frame,)) diff --git a/pyjelly/parse/lookup.py b/pyjelly/parse/lookup.py index 94cc56fe..e9aea12e 100644 --- a/pyjelly/parse/lookup.py +++ b/pyjelly/parse/lookup.py @@ -44,10 +44,10 @@ def assign_entry(self, index: int, value: str) -> None: def at(self, index: int) -> str: self.last_reused_index = index - value = self.data[index - 1] + value = self.data[index - 1] if 0 < index <= self.lookup_size else None if value is None: - msg = f"invalid resolved index {index}" - raise IndexError(msg) + msg = f"a term refers to lookup entry {index}, which is not set" + raise JellyConformanceError(msg) return value def decode_prefix_term_index(self, index: int) -> str: diff --git a/pyjelly/parse/sparql.py b/pyjelly/parse/sparql.py index f0180826..2d7d9acf 100644 --- a/pyjelly/parse/sparql.py +++ b/pyjelly/parse/sparql.py @@ -12,8 +12,7 @@ from collections.abc import Callable, Iterator from typing import IO, Any, Final -from google.protobuf.message import DecodeError -from google.protobuf.proto import parse, parse_length_prefixed +from google.protobuf.proto import parse from pyjelly import jelly from pyjelly.errors import ( @@ -36,7 +35,12 @@ ) from pyjelly.parse.columns import MAX_ROWS, ColumnDecoder from pyjelly.parse.decode import Adapter, Decoder, ParserOptions -from pyjelly.parse.ioutils import SPARQL_FRAME_TAGS, detect_delimited +from pyjelly.parse.ioutils import ( + DECODE_ERRORS, + SPARQL_FRAME_TAGS, + detect_delimited, + parse_delimited, +) #: Metadata key for the links of a result set (head.link in SPARQL JSON results) LINK_KEY: Final = "link" @@ -58,10 +62,6 @@ def tuple_rows(variables: list[str], cells: list[list[Any]], rows: int) -> list[ # every attribute name of a message as a field name first _FRAME_HAS_FIELD: Final = jelly.SparqlResultsFrame.HasField -# What protobuf raises for bytes it can't decode. The pure-Python implementation -# raises UnicodeDecodeError for a string that isn't UTF-8, the others DecodeError. -_DECODE_ERRORS: Final = (DecodeError, UnicodeDecodeError) - _STREAM_TYPES: Final = ( jelly.SPARQL_STREAM_TYPE_FLAT, @@ -411,7 +411,7 @@ def sparql_frames(inp: IO[bytes] | bytes) -> Iterator[jelly.SparqlResultsFrame]: raise JellyConformanceError(msg) try: return iter((parse(jelly.SparqlResultsFrame, data),)) - except _DECODE_ERRORS as error: + except DECODE_ERRORS as error: raise _not_sparql(error) from None return _delimited_frames(inp) @@ -427,18 +427,16 @@ def _not_sparql(error: Exception) -> JellyConformanceError: def _delimited_frames(inp: IO[bytes]) -> Iterator[jelly.SparqlResultsFrame]: while True: try: - # protobuf annotates `input_bytes` as io.BytesIO, but any IO[bytes] works. - frame = parse_length_prefixed(jelly.SparqlResultsFrame, inp) # type: ignore[arg-type] - except _DECODE_ERRORS as error: + frame = parse_delimited(jelly.SparqlResultsFrame, inp) + except DECODE_ERRORS as error: raise _not_sparql(error) from None except ValueError as error: - if getattr(inp, "closed", False): - msg = ( - "the input was closed before all results were read. Read the " - "rows before closing it." - ) - raise ValueError(msg) from error - raise + # parse_delimited() raises it only if the input is closed + msg = ( + "the input was closed before all results were read. Read the rows " + "before closing it." + ) + raise ValueError(msg) from error if frame is None: return yield frame diff --git a/tests/unit_tests/test_parse/test_decoder.py b/tests/unit_tests/test_parse/test_decoder.py index f45f73bf..adb5531a 100644 --- a/tests/unit_tests/test_parse/test_decoder.py +++ b/tests/unit_tests/test_parse/test_decoder.py @@ -1,6 +1,7 @@ import pytest from pyjelly import jelly +from pyjelly.errors import JellyConformanceError from pyjelly.integrations.generic.decode import GenericTriplesAdapter from pyjelly.integrations.generic.generic_sink import IRI, Literal from pyjelly.options import VERSION_1_1, LookupPreset, StreamParameters, StreamTypes @@ -66,7 +67,7 @@ def test_repeated_terms_must_be_set_first() -> None: decoder.ingest_prefix_entry(jelly.RdfPrefixEntry(id=1, value="http://a/")) decoder.ingest_name_entry(jelly.RdfNameEntry(id=1, value="x")) iri = jelly.RdfIri(prefix_id=1, name_id=1) - with pytest.raises(ValueError, match="missing repeated term predicate"): + with pytest.raises(JellyConformanceError, match="missing repeated term predicate"): decoder.decode_triple(jelly.RdfTriple(s_iri=iri, o_iri=iri)) # Terms set once can be repeated @@ -80,5 +81,5 @@ def test_repeated_terms_must_be_set_first() -> None: def test_empty_row_is_an_error() -> None: frame = jelly.RdfStreamFrame(rows=[jelly.RdfStreamRow()]) - with pytest.raises(TypeError, match="decoder not implemented for row None"): + with pytest.raises(JellyConformanceError, match="a row of the stream is empty"): list(make_decoder().iter_rows(frame)) diff --git a/tests/unit_tests/test_parse/test_ioutils.py b/tests/unit_tests/test_parse/test_ioutils.py index ccb5c6ff..278a2ef9 100644 --- a/tests/unit_tests/test_parse/test_ioutils.py +++ b/tests/unit_tests/test_parse/test_ioutils.py @@ -9,15 +9,24 @@ from google.protobuf.descriptor import Descriptor from pyjelly import jelly +from pyjelly.integrations.generic.decode import GenericStatementSinkAdapter +from pyjelly.integrations.generic.generic_sink import IRI from pyjelly.parse.ioutils import ( FRAME_TAGS, + SPARQL_FRAME_TAGS, delimited_jelly_hint, get_options_and_frames, read_hint_header, ) +from pyjelly.parse.sparql import read_result_set from tests.utils import wire_fuzzer FRAME_DESCRIPTOR = cast("Descriptor", jelly.RdfStreamFrame.DESCRIPTOR) +SPARQL_FRAME_DESCRIPTOR = cast("Descriptor", jelly.SparqlResultsFrame.DESCRIPTOR) +FRAME_TYPES = { + "rdf": (FRAME_DESCRIPTOR, FRAME_TAGS), + "sparql": (SPARQL_FRAME_DESCRIPTOR, SPARQL_FRAME_TAGS), +} # A frame of 25 bytes: too long for its first byte to be a size. FRAME_LARGE = jelly.RdfStreamFrame( @@ -37,11 +46,11 @@ ) -def detect(data: bytes) -> tuple[bool, bytes]: +def detect(data: bytes, tags: frozenset[int] = FRAME_TAGS) -> tuple[bool, bytes]: """Run the detection and return its answer plus every byte of the input.""" inp = io.BytesIO(data) - header = read_hint_header(inp) - return delimited_jelly_hint(header), header + inp.read() + header = read_hint_header(inp, tags) + return delimited_jelly_hint(header, tags), header + inp.read() def delimit(*frames: jelly.RdfStreamFrame) -> bytes: @@ -154,15 +163,17 @@ def read(self, size: int = -1) -> bytes: # noqa: ARG002 ITERATIONS = 5_000 -def test_fuzz_detects_random_non_delimited_frames() -> None: +@pytest.mark.parametrize("frame_type", list(FRAME_TYPES)) +def test_fuzz_detects_random_non_delimited_frames(frame_type: str) -> None: + descriptor, tags = FRAME_TYPES[frame_type] rnd = random.Random(1) failures = 0 ambiguous = 0 for _ in range(ITERATIONS): - data = wire_fuzzer.message(FRAME_DESCRIPTOR, rnd) - if data and data[0] in FRAME_TAGS: + data = wire_fuzzer.message(descriptor, rnd) + if data and data[0] in tags: ambiguous += 1 - is_delimited, kept = detect(data) + is_delimited, kept = detect(data, tags) assert kept == data failures += is_delimited # A frame that does not start with the stream options can, by chance, also be a @@ -172,30 +183,86 @@ def test_fuzz_detects_random_non_delimited_frames() -> None: assert ambiguous > ITERATIONS // 2 -def test_fuzz_detects_random_delimited_streams() -> None: +@pytest.mark.parametrize("frame_type", list(FRAME_TYPES)) +def test_fuzz_detects_random_delimited_streams(frame_type: str) -> None: + descriptor, tags = FRAME_TYPES[frame_type] rnd = random.Random(2) ambiguous = 0 for _ in range(ITERATIONS): data = b"" for _ in range(rnd.randrange(1, 4)): - body = wire_fuzzer.message(FRAME_DESCRIPTOR, rnd) + body = wire_fuzzer.message(descriptor, rnd) data += wire_fuzzer.varint(len(body)) + body - if data[0] in FRAME_TAGS: + if data[0] in tags: ambiguous += 1 - assert detect(data) == (True, data) - # The first frame is sometimes 10, 18 or 122 bytes long: sizes that are also tags + assert detect(data, tags) == (True, data) + # Some first frames have a length that is also a tag byte assert ambiguous > 0 -def test_fuzz_keeps_every_byte_of_random_input() -> None: +@pytest.mark.parametrize("frame_type", list(FRAME_TYPES)) +def test_fuzz_keeps_every_byte_of_random_input(frame_type: str) -> None: + _, tags = FRAME_TYPES[frame_type] rnd = random.Random(3) for _ in range(ITERATIONS): length = rnd.randrange(400) if rnd.randrange(10) == 0 else rnd.randrange(24) data = bytes(rnd.randrange(256) for _ in range(length)) - _, kept = detect(data) + _, kept = detect(data, tags) assert kept == data +# Jelly-SPARQL stream options of several serialized sizes +SPARQL_OPTIONS = [ + jelly.SparqlResultsOptions( + max_name_table_size=names, + max_prefix_table_size=prefixes, + max_datatype_table_size=64, + version=1, + ) + for names in (128, 256, 1024, 8192, 16384) + for prefixes in (0, 64, 1024, 4096) +] + + +def sparql_frame(options: jelly.SparqlResultsOptions) -> bytes: + return jelly.SparqlResultsFrame( + options=options, + variables=["x"], + row_count=1, + names=[jelly.RdfLookupEntryPacked(values=["http://example.org/a"])], + columns=[jelly.RdfColumn(name_ids=[1])], + trailer=jelly.SparqlResultsTrailer(), + ).SerializeToString() + + +def test_sparql_options_cover_the_size_of_a_tag() -> None: + # Options of 10 bytes make a non-delimited frame start with 0A 0A, which is + # also how a delimited stream whose first frame is 10 bytes long starts + assert 10 in {options.ByteSize() for options in SPARQL_OPTIONS} + + +@pytest.mark.parametrize( + "options", + SPARQL_OPTIONS, + ids=[ + f"names={o.max_name_table_size},prefixes={o.max_prefix_table_size}" + for o in SPARQL_OPTIONS + ], +) +@pytest.mark.parametrize("delimited", [True, False], ids=["delimited", "non-delimited"]) +def test_detects_sparql_frames( + options: jelly.SparqlResultsOptions, + delimited: bool, # noqa: FBT001 +) -> None: + data = sparql_frame(options) + if delimited: + data = wire_fuzzer.varint(len(data)) + data + assert detect(data, SPARQL_FRAME_TAGS) == (delimited, data) + # Read it back, delimited or not as detected + result_set = read_result_set(data, GenericStatementSinkAdapter) + assert list(result_set) == [(IRI("http://example.org/a"),)] + + # A frame holding only the stream options. OPTIONS_FRAME = b"\x0a\x0f\x0a\x0d\x10\x01\x48\x80\x01\x50\x0f\x58\x0f\x70\x01\x78\x01" diff --git a/tests/unit_tests/test_parse/test_lookup_decoder.py b/tests/unit_tests/test_parse/test_lookup_decoder.py index ab1b0568..d69fe8ff 100644 --- a/tests/unit_tests/test_parse/test_lookup_decoder.py +++ b/tests/unit_tests/test_parse/test_lookup_decoder.py @@ -20,12 +20,19 @@ def test_datatype_index_zero_error() -> None: def test_at_invalid_index() -> None: decoder = LookupDecoder(lookup_size=4) - with pytest.raises(IndexError) as excinfo: + with pytest.raises(JellyConformanceError, match="lookup entry 2, which is not set"): decoder.at(2) - assert "invalid resolved index 2" in str(excinfo.value) assert decoder.last_reused_index == 2 +@pytest.mark.parametrize("index", [0, 5, 2**32 - 1]) +def test_at_outside_lookup(index: int) -> None: + decoder = LookupDecoder(lookup_size=4) + decoder.assign_entry(4, "d") + with pytest.raises(JellyConformanceError, match="which is not set"): + decoder.at(index) + + def test_assign_entry_outside_lookup() -> None: decoder = LookupDecoder(lookup_size=2) decoder.assign_entry(2, "b") diff --git a/tests/unit_tests/test_parse/test_rdf_hardening.py b/tests/unit_tests/test_parse/test_rdf_hardening.py new file mode 100644 index 00000000..920b4a99 --- /dev/null +++ b/tests/unit_tests/test_parse/test_rdf_hardening.py @@ -0,0 +1,355 @@ +""" +Reading invalid Jelly-RDF input, after jelly-jvm's ProtoDecoderHardeningSpec. + +The rows are valid protobuf, but refer to things that are not there or are not +allowed in the stream. The reader must reject them with JellyConformanceError. +""" + +from __future__ import annotations + +import io +from collections.abc import Callable +from typing import IO, Any + +import pytest + +from pyjelly import jelly +from pyjelly.errors import JellyAssertionError, JellyConformanceError +from pyjelly.integrations.generic.decode import GenericStatementSinkAdapter +from pyjelly.integrations.generic.generic_sink import ( + IRI, + BlankNode, + DefaultGraph, + GenericStatementSink, + Literal, + Quad, + Triple, +) +from pyjelly.integrations.generic.parse import parse_jelly_flat +from pyjelly.integrations.generic.serialize import guess_stream, stream_frames +from pyjelly.options import LookupPreset, StreamParameters +from pyjelly.parse.sparql import read_result_sets +from pyjelly.serialize.ioutils import write_delimited, write_single +from pyjelly.serialize.streams import SerializerOptions +from tests.utils import byte_fuzzer, wire_fuzzer + +Row = jelly.RdfStreamRow +DATATYPES = 8 + + +def options(datatypes: int = DATATYPES) -> jelly.RdfStreamRow: + return Row( + options=jelly.RdfStreamOptions( + physical_type=jelly.PHYSICAL_STREAM_TYPE_TRIPLES, + max_name_table_size=128, + max_prefix_table_size=8, + max_datatype_table_size=datatypes, + version=2, + ) + ) + + +def iri(prefix: int, name: int) -> jelly.RdfIri: + return jelly.RdfIri(prefix_id=prefix, name_id=name) + + +def name(index: int, value: str) -> jelly.RdfStreamRow: + return Row(name=jelly.RdfNameEntry(id=index, value=value)) + + +def prefix(index: int, value: str) -> jelly.RdfStreamRow: + return Row(prefix=jelly.RdfPrefixEntry(id=index, value=value)) + + +def datatype(index: int, value: str) -> jelly.RdfStreamRow: + return Row(datatype=jelly.RdfDatatypeEntry(id=index, value=value)) + + +def triple(**terms: Any) -> jelly.RdfStreamRow: + return Row(triple=jelly.RdfTriple(**terms)) + + +def namespace(label: str, value: jelly.RdfIri | None = None) -> jelly.RdfStreamRow: + return Row(namespace=jelly.RdfNamespaceDeclaration(name=label, value=value)) + + +def read(data: bytes) -> list[Any]: + return list(parse_jelly_flat(io.BytesIO(data))) + + +def read_rows(*rows: jelly.RdfStreamRow) -> list[Any]: + out = io.BytesIO() + write_delimited(jelly.RdfStreamFrame(rows=rows), out) + return read(out.getvalue()) + + +def literal(lex: str, index: int) -> jelly.RdfStreamRow: + """Return the triple "lex"^^dt, with dt the datatype at the index.""" + return triple( + s_iri=iri(0, 1), + p_iri=iri(0, 1), + o_literal=jelly.RdfLiteral(lex=lex, datatype=index), + ) + + +REJECTED = { + "datatype entry past the end": ( + [options(), datatype(DATATYPES + 1, "http://dt")], + "the lookup has 8 entries", + ), + "datatype entry id above 2^31": ( + [options(), datatype(2**32 - 1, "http://dt")], + "the lookup has 8 entries", + ), + "datatype entry with no datatype lookup": ( + [options(datatypes=0), datatype(1, "http://dt")], + "the lookup has 0 entries", + ), + "datatype entries that run off the end": ( + [options()] + [datatype(0, f"http://dt{i}") for i in range(DATATYPES + 1)], + "the lookup has 8 entries", + ), + "literal with a datatype never set": ( + [options(), name(1, "http://s"), literal("1", 1)], + "lookup entry 1, which is not set", + ), + "literal with a datatype past the end": ( + [options(), name(1, "http://s"), literal("1", DATATYPES + 1)], + "lookup entry 9, which is not set", + ), + "literal with a datatype, with no datatype lookup": ( + [options(datatypes=0), name(1, "http://s"), literal("1", 1)], + "lookup entry 1, which is not set", + ), + "namespace declaration with no IRI": ( + [options(), namespace("ex")], + "which is not set", + ), + "namespace declaration with a name never set": ( + [options(), prefix(1, "http://example.org/"), namespace("ex", iri(1, 1))], + "lookup entry 1, which is not set", + ), + "namespace declaration with a prefix never set": ( + [options(), name(1, "x"), namespace("ex", iri(1, 1))], + "lookup entry 1, which is not set", + ), + "IRI with a name id above 2^31": ( + [options(), triple(s_iri=iri(0, 2**32 - 1), p_iri=iri(0, 1), o_iri=iri(0, 1))], + "lookup entry 4294967295, which is not set", + ), + "IRI with a prefix id above 2^31": ( + [ + options(), + name(1, "x"), + triple(s_iri=iri(2**32 - 1, 1), p_iri=iri(0, 1), o_iri=iri(0, 1)), + ], + "lookup entry 4294967295, which is not set", + ), + "row with nothing in it": ([options(), Row()], "a row of the stream is empty"), + "repeated stream options that are not the same": ( + [options(), options(datatypes=16)], + "not the same as the first ones", + ), + "quads in a stream of triples": ( + [ + options(), + name(1, "http://s"), + Row( + quad=jelly.RdfQuad( + s_iri=iri(0, 1), p_iri=iri(0, 1), o_iri=iri(0, 1), g_iri=iri(0, 1) + ) + ), + ], + "PHYSICAL_STREAM_TYPE_TRIPLES cannot have quads", + ), + "graph start in a stream of triples": ( + [options(), Row(graph_start=jelly.RdfGraphStart(g_default_graph={}))], + "PHYSICAL_STREAM_TYPE_TRIPLES cannot have graph start markers", + ), + "graph start with no graph": ( + [options(), Row(graph_start=jelly.RdfGraphStart())], + "a graph start marker has no graph", + ), + "quoted triple with a repeated term": ( + [ + options(), + name(1, "http://s"), + triple( + s_iri=iri(0, 1), + p_iri=iri(0, 1), + o_triple_term=jelly.RdfTriple(s_iri=iri(0, 1), p_iri=iri(0, 1)), + ), + ], + "repeated terms are not allowed in quoted triples", + ), + "repeated term with no term before it": ( + [options(), triple()], + "missing repeated term subject", + ), +} + + +@pytest.mark.parametrize(("rows", "error"), list(REJECTED.values()), ids=list(REJECTED)) +def test_rejects(rows: list[jelly.RdfStreamRow], error: str) -> None: + with pytest.raises(JellyConformanceError, match=error): + read_rows(*rows) + + +def test_rejects_namespace_declaration_with_a_name_never_set_in_columns() -> None: + batch = jelly.RdfColumnBatch( + prefixes=[jelly.RdfLookupEntryPacked(values=["http://example.org/"])], + namespaces=[jelly.RdfNamespaceDeclaration(name="ex", value=iri(1, 1))], + ) + stream_options = jelly.RdfStreamOptions( + physical_type=jelly.PHYSICAL_STREAM_TYPE_TRIPLES, + max_name_table_size=128, + max_prefix_table_size=8, + version=3, + ) + out = io.BytesIO() + write_delimited( + jelly.RdfStreamFrame(rows=[Row(options=stream_options)], columns=batch), out + ) + with pytest.raises(JellyConformanceError, match="which is not set"): + read(out.getvalue()) + + +# Length prefixes of delimited streams, after jelly-jvm's DelimitedMessageReaderSpec + +RDF_FRAME = jelly.RdfStreamFrame( + rows=[options(), name(1, "http://example.org/" + "x" * 3000)] +).SerializeToString() +SPARQL_FRAME = jelly.SparqlResultsFrame( + options=jelly.SparqlResultsOptions(max_name_table_size=128, version=1), + variables=["x"], + row_count=1, + names=[jelly.RdfLookupEntryPacked(values=["http://example.org/" + "x" * 3000])], + columns=[jelly.RdfColumn(name_ids=[1])], +).SerializeToString() + + +def read_rdf(inp: IO[bytes]) -> None: + list(parse_jelly_flat(inp)) + + +def read_sparql(inp: IO[bytes]) -> None: + for result_set in read_result_sets(inp, GenericStatementSinkAdapter): + list(result_set) + + +READERS = {"rdf": (RDF_FRAME, read_rdf), "sparql": (SPARQL_FRAME, read_sparql)} + +BAD_LENGTHS: dict[str, Callable[[bytes], bytes]] = { + "frame shorter than its length, cut at 3": lambda delimited: delimited[:3], + "frame shorter than its length, cut at 100": lambda delimited: delimited[:100], + "frame shorter than its length by 1": lambda delimited: delimited[:-1], + "length of 2^31 - 1": lambda _: b"\xff\xff\xff\xff\x07\x01\x02\x03", + "length of 2^32 - 1": lambda _: b"\xff\xff\xff\xff\x0f\x01\x02\x03", + "length of 2^64 - 1": lambda _: b"\xff" * 9 + b"\x01\x01\x02", + "length cut off": lambda _: b"\x80", + "length cut off after a frame": lambda delimited: delimited + b"\x80", + "frame cut off after a frame": lambda delimited: delimited + delimited[:50], +} + + +@pytest.mark.parametrize("reader", list(READERS)) +@pytest.mark.parametrize("bad", list(BAD_LENGTHS)) +def test_rejects_bad_length(reader: str, bad: str) -> None: + frame, read_stream = READERS[reader] + delimited = wire_fuzzer.varint(len(frame)) + frame + with pytest.raises(JellyConformanceError, match="Is it cut short"): + read_stream(io.BytesIO(BAD_LENGTHS[bad](delimited))) + + +@pytest.mark.parametrize("reader", list(READERS)) +def test_closed_input(reader: str) -> None: + frame, read_stream = READERS[reader] + inp = io.BytesIO(wire_fuzzer.varint(len(frame)) + frame) + inp.close() + with pytest.raises(ValueError, match="closed"): + read_stream(inp) + + +def corpus() -> list[bytes]: + """ + Return valid streams to start the fuzzing from. + + Triples and quads in the row layout (Jelly-RDF 1.1) and the column layout + (1.2), delimited and not. They have every term type, lookup entries, repeated + terms, namespace declarations and quoted triples. + """ + ex = "http://example.org/" + s, p, o = IRI(ex + "s"), IRI(ex + "p"), IRI(ex + "o") + triples = [ + Triple(s, p, o), + Triple(s, p, Literal("plain")), + Triple(s, IRI(ex + "q"), Literal("1", datatype="http://dt/int")), + Triple(BlankNode("b1"), p, Literal("hello", langtag="en")), + Triple(BlankNode("b1"), p, Triple(s, p, o)), + Triple(s, p, o), + ] + graph = IRI(ex + "g") + quads = [ + Quad(t.s, t.p, t.o, graph if i % 2 else DefaultGraph) + for i, t in enumerate(triples) + ] + streams = [] + for statements, logical_type in ( + (triples, jelly.LOGICAL_STREAM_TYPE_FLAT_TRIPLES), + (quads, jelly.LOGICAL_STREAM_TYPE_FLAT_QUADS), + ): + for version, frame_size in ((2, 4), (2, 250), (3, 4), (3, 250)): + sink = GenericStatementSink() + sink.bind("ex", IRI(ex)) + for statement in statements: + sink.add(statement) + params = StreamParameters( + version=version, + rdf_star=version == 2, + namespace_declarations=True, + delimited=frame_size == 4, + ) + serializer_options = SerializerOptions( + frame_size=frame_size, + logical_type=logical_type, + params=params, + lookup_preset=LookupPreset( + max_names=128, max_prefixes=8, max_datatypes=8 + ), + ) + out = io.BytesIO() + for frame in stream_frames(guess_stream(serializer_options, sink), sink): + if params.delimited: + write_delimited(frame, out) + else: + write_single(frame, out) + streams.append(out.getvalue()) + return streams + + +def test_corpus_is_valid() -> None: + for data in corpus(): + assert len([s for s in read(data) if isinstance(s, (Triple, Quad))]) == 6 + + +def test_fuzz_mutated_streams() -> None: + reached = 0 + + def consume(data: bytes) -> None: + nonlocal reached + try: + read(data) + except JellyConformanceError as error: + if "Is it cut short" not in str(error): + reached += 1 + raise + reached += 1 + + # Options whose physical and logical types don't match raise + # JellyAssertionError + found = byte_fuzzer.findings( + corpus(), consume, (JellyConformanceError, JellyAssertionError) + ) + assert not found, "\n".join(found) + # Most mutations are rejected by protobuf. At least 5% must get to the decoder + assert reached > byte_fuzzer.ITERATIONS // 20 diff --git a/tests/unit_tests/test_parse/test_sparql_hardening.py b/tests/unit_tests/test_parse/test_sparql_hardening.py new file mode 100644 index 00000000..d9c2e176 --- /dev/null +++ b/tests/unit_tests/test_parse/test_sparql_hardening.py @@ -0,0 +1,399 @@ +""" +Reading invalid Jelly-SPARQL input, after jelly-jvm's SparqlDecoderHardeningSpec. + +The frames are valid protobuf, but declare huge sizes or refer to things that are +not there. The reader must reject them with JellyConformanceError, without +allocating memory for the declared sizes. +""" + +from __future__ import annotations + +import io +import warnings +from typing import Any + +import pytest + +from pyjelly import jelly +from pyjelly.errors import ( + JellyConformanceError, + JellyIncompleteResultsError, + JellyTruncatedResultsWarning, +) +from pyjelly.integrations.generic import sparql as generic_sparql +from pyjelly.integrations.generic.decode import GenericStatementSinkAdapter +from pyjelly.integrations.generic.generic_sink import IRI, BlankNode, Literal +from pyjelly.options import ( + MAX_SPARQL_FRAME_ROWS, + MAX_SPARQL_FRAME_VALUES, + LookupPreset, + SparqlOptions, +) +from pyjelly.parse.sparql import SparqlDecoder, read_result_sets +from tests.utils import byte_fuzzer + +OPTIONS = jelly.SparqlResultsOptions( + max_name_table_size=128, + max_prefix_table_size=16, + max_datatype_table_size=16, + version=1, +) +DATATYPES = OPTIONS.max_datatype_table_size + +Entry = jelly.RdfLookupEntryPacked +Column = jelly.RdfColumn + +KIND_IRI = 0 +KIND_LITERAL = 1 +KIND_BNODE = 2 + + +def decoder( + *, max_rows: int = MAX_SPARQL_FRAME_ROWS, max_values: int = MAX_SPARQL_FRAME_VALUES +) -> SparqlDecoder: + return SparqlDecoder( + GenericStatementSinkAdapter, max_rows=max_rows, max_values=max_values + ) + + +def frame( + rows: int, *columns: jelly.RdfColumn, **fields: Any +) -> jelly.SparqlResultsFrame: + """Return a frame with the stream options and the variable x, unless given.""" + fields.setdefault("variables", ["x"]) + return jelly.SparqlResultsFrame( + options=OPTIONS, row_count=rows, columns=list(columns), **fields + ) + + +def kinds(values: list[int]) -> bytes: + """Pack the kinds of the values, 2 bits each.""" + out = bytearray((len(values) + 3) // 4) + for i, kind in enumerate(values): + out[i >> 2] |= kind << ((i & 3) << 1) + return bytes(out) + + +def datatype(index: int) -> int: + """Return the literal kind of a literal with this datatype.""" + return 2 * index - 1 + + +def langtag(index: int) -> int: + """Return the literal kind of a literal with this language tag.""" + return 2 * index + 2 + + +def decode(data: jelly.SparqlResultsFrame, **limits: int) -> list[Any]: + return decoder(**limits).decode(data).rows + + +REJECTED = { + "row count above 2^27": (frame(1 << 27, variables=[]), "at most 134217727 rows"), + "row count above 2^31": (frame(2**31 - 1, Column()), "at most 134217727 rows"), + "row count above 2^32 - 1": ( + frame(2**32 - 1, variables=[]), + "not 4294967295", + ), + "row count above the default limit": ( + frame(MAX_SPARQL_FRAME_ROWS + 1, variables=[]), + f"more than the {MAX_SPARQL_FRAME_ROWS} allowed", + ), + "datatype entry past the end": ( + frame(0, Column(), datatypes=[Entry(id=DATATYPES + 1, values=["http://dt"])]), + "the lookup has 16 entries", + ), + "datatype entry id above 2^31": ( + frame(0, Column(), datatypes=[Entry(id=2**32 - 1, values=["http://dt"])]), + "the lookup has 16 entries", + ), + "datatype entries that run off the end": ( + frame( + 0, + Column(), + datatypes=[ + Entry(id=1, values=[f"http://dt{i}" for i in range(DATATYPES + 1)]) + ], + ), + "the lookup has 16 entries", + ), + "literal with a datatype past the end": ( + frame(1, Column(lex_values=["1"], literal_kinds=[datatype(DATATYPES + 1)])), + "outside the datatype lookup", + ), + "literal with a datatype never set": ( + frame(1, Column(lex_values=["1"], literal_kinds=[datatype(1)])), + "lookup entry 1, which is not set", + ), + "literal kind per value with a datatype never set": ( + frame(2, Column(lex_values=["1", "2"], literal_kinds=[0, datatype(1)])), + "lookup entry 1, which is not set", + ), + "literal with a language tag the column does not have": ( + frame( + 2, + Column( + lex_values=["a", "b"], + literal_kinds=[langtag(0), langtag(1)], + langtags=["en"], + ), + ), + "language tag 1", + ), + "uniform literal kind with no language tags": ( + frame(1, Column(lex_values=["a"], literal_kinds=[langtag(0)])), + "language tag 0", + ), + "more than one literal kind, but not one per literal": ( + frame(3, Column(lex_values=["a", "b", "c"], literal_kinds=[0, 0])), + "2 literal kinds for 3 literals", + ), + "base directions for only some language tags": ( + frame( + 2, + Column( + lex_values=["a", "b"], + literal_kinds=[langtag(0), langtag(1)], + langtags=["en", "fr"], + langtag_directions=[jelly.RDF_BASE_DIRECTION_LTR], + ), + ), + "1 base directions for 2 language tags", + ), + "unknown base direction": ( + frame( + 1, + Column( + lex_values=["a"], + literal_kinds=[langtag(0)], + langtags=["en"], + langtag_directions=[7], + ), + ), + "unknown base direction: 7", + ), + "kinds of the wrong length": ( + frame(2, Column(kinds=b"\x06\x00", bnodes=["a"], lex_values=["x"])), + "2 bytes of kinds for 2 values", + ), + "kinds that do not match the values": ( + frame( + 2, + Column( + kinds=kinds([KIND_BNODE, KIND_BNODE]), bnodes=["a"], lex_values=["x"] + ), + ), + "more blank nodes than it has", + ), + "kinds with unused bits set": ( + frame(2, Column(kinds=bytes([0x06 | 0x40]), bnodes=["a"], lex_values=["x"])), + "unused bits", + ), + "IRI with a name never set": ( + frame( + 1, + Column(name_ids=[1], prefix_ids=[1]), + prefixes=[Entry(id=1, values=["http://example.org/"])], + ), + "name 1 or prefix 1, which is not set", + ), + "IRI with a prefix never set": ( + frame( + 1, + Column(name_ids=[1], prefix_ids=[1]), + names=[Entry(id=1, values=["x"])], + ), + "name 1 or prefix 1, which is not set", + ), + "IRI with kinds, with a name never set": ( + frame( + 1, + Column(kinds=kinds([KIND_IRI]), name_ids=[1], prefix_ids=[1]), + prefixes=[Entry(id=1, values=["http://example.org/"])], + ), + "name 1 or prefix 1, which is not set", + ), + "IRI with a name id above 2^31": ( + frame(1, Column(name_ids=[2**32 - 1])), + "outside the lookups", + ), + "IRI with a prefix id above 2^31": ( + frame( + 1, + Column(name_ids=[1], prefix_ids=[2**32 - 1]), + names=[Entry(id=1, values=["x"])], + ), + "outside the lookups", + ), +} + + +@pytest.mark.parametrize(("data", "error"), list(REJECTED.values()), ids=list(REJECTED)) +def test_rejects(data: jelly.SparqlResultsFrame, error: str) -> None: + with pytest.raises(JellyConformanceError, match=error): + decode(data) + + +def test_row_count_up_to_the_limit() -> None: + data = frame( + 5, Column(name_ids=[1]), names=[Entry(id=1, values=["http://example.org/x"])] + ) + rows = decode(data, max_rows=5) + assert rows == [(IRI("http://example.org/x"),)] + [(None,)] * 4 + with pytest.raises(JellyConformanceError, match="more than the 4 allowed"): + decode(data, max_rows=4) + + +def test_value_count_above_the_default_limit() -> None: + # 1000 empty columns of 2^20 rows each: a few kilobytes of input that would + # need gigabytes of cells. The row count is within the row limit. + variables = [f"v{i}" for i in range(1000)] + data = frame( + MAX_SPARQL_FRAME_ROWS, *[Column()] * len(variables), variables=variables + ) + with pytest.raises( + JellyConformanceError, match=f"the {MAX_SPARQL_FRAME_VALUES} values allowed" + ): + decode(data) + + +def test_value_count_up_to_the_limit() -> None: + def data(rows: int) -> jelly.SparqlResultsFrame: + return frame(rows, Column(), Column(), variables=["x", "y"]) + + assert len(decode(data(5), max_values=10)) == 5 + with pytest.raises(JellyConformanceError, match="6 rows of 2 variables"): + decode(data(6), max_values=10) + + +# A column long enough to check that the kinds are read to the end +LONG_COLUMN = 70_001 + + +def long_mixed_column() -> tuple[jelly.RdfColumn, list[int]]: + """Return a column of blank nodes and literals, and its kinds.""" + values = [KIND_BNODE if i % 3 == 0 else KIND_LITERAL for i in range(LONG_COLUMN)] + column = Column( + kinds=kinds(values), + bnodes=[f"b{i}" for i in range(0, LONG_COLUMN, 3)], + lex_values=[f"l{i}" for i in range(LONG_COLUMN) if i % 3], + ) + return column, values + + +def test_long_mixed_column() -> None: + column, _ = long_mixed_column() + rows = decode(frame(LONG_COLUMN, column)) + assert len(rows) == LONG_COLUMN + for i in (0, 1, 65531, 65532, 65533, 70000): + expected = BlankNode(f"b{i}") if i % 3 == 0 else Literal(f"l{i}") + assert str(rows[i][0]) == str(expected) + + +def test_long_mixed_column_with_a_mismatch_near_the_end() -> None: + column, values = long_mixed_column() + # Value 69998 is a literal, the kinds say it is a blank node + values[69998] = KIND_BNODE + column.kinds = kinds(values) + with pytest.raises(JellyConformanceError, match="more blank nodes than it has"): + decode(frame(LONG_COLUMN, column)) + + +def test_kinds_on_a_column_of_one_type() -> None: + column = Column(kinds=kinds([KIND_BNODE, KIND_BNODE]), bnodes=["a", "b"]) + rows = decode(frame(2, column)) + assert [str(row[0]) for row in rows] == ["_:a", "_:b"] + + +def test_base_direction_checked_only_where_used() -> None: + # The second tag has an unknown direction, but no value refers to it + column = Column( + lex_values=["a"], + literal_kinds=[langtag(0)], + langtags=["en", "fr"], + langtag_directions=[jelly.RDF_BASE_DIRECTION_UNSPECIFIED, 7], + ) + assert decode(frame(1, column)) == [(Literal("a", langtag="en"),)] + + +def test_column_with_no_values() -> None: + assert decode(frame(1, Column())) == [(None,)] + + +def corpus() -> list[bytes]: + """ + Return valid streams to start the fuzzing from. + + They have every term type, columns with and without kinds, lookup entries, + repeated and unbound cells, delimited and non-delimited streams, a trailer + and a boolean result. + """ + a = IRI("http://example.org/a") + rows: list[tuple[Any, ...]] = [ + (a, Literal("plain"), BlankNode("b1")), + # Repeats the row before it + (a, Literal("plain"), BlankNode("b1")), + (IRI("http://example.org/b"), Literal("1", datatype="http://dt/int"), None), + # Mixes term types in the columns, so that they get kinds + (BlankNode("b2"), Literal("hello", langtag="en"), IRI("http://example.org/c")), + (None, None, None), + ( + IRI("http://example.org/d"), + Literal("x", langtag="ar", direction="rtl"), + BlankNode("b3"), + ), + ] + small = LookupPreset(max_names=128, max_prefixes=8, max_datatypes=8) + streams = [] + for options in ( + SparqlOptions(lookup_preset=small, delimited=False), + SparqlOptions(lookup_preset=small, frame_values=6), + ): + out = io.BytesIO() + generic_sparql.serialize_jelly_sparql(["x", "y", "z"], rows, out, options) + streams.append(out.getvalue()) + out = io.BytesIO() + generic_sparql.serialize_jelly_sparql_ask( + value=True, out=out, options=SparqlOptions(lookup_preset=small, delimited=False) + ) + streams.append(out.getvalue()) + return streams + + +def read_all(data: bytes) -> None: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", JellyTruncatedResultsWarning) + # A low row limit, so that mutated row counts are rejected quickly + for result_set in read_result_sets( + data, GenericStatementSinkAdapter, max_rows=64 + ): + if result_set.is_boolean: + result_set.boolean # noqa: B018 + else: + list(result_set) + + +def test_corpus_is_valid() -> None: + for data in corpus(): + read_all(data) + + +def test_fuzz_mutated_streams() -> None: + reached = 0 + + def consume(data: bytes) -> None: + nonlocal reached + try: + read_all(data) + except JellyConformanceError as error: + if "Is it cut short" not in str(error): + reached += 1 + raise + reached += 1 + + found = byte_fuzzer.findings( + corpus(), consume, (JellyConformanceError, JellyIncompleteResultsError) + ) + assert not found, "\n".join(found) + # Most mutations are rejected by protobuf. At least 5% must get to the decoder + assert reached > byte_fuzzer.ITERATIONS // 20 diff --git a/tests/utils/byte_fuzzer.py b/tests/utils/byte_fuzzer.py new file mode 100644 index 00000000..4245ab62 --- /dev/null +++ b/tests/utils/byte_fuzzer.py @@ -0,0 +1,107 @@ +""" +Byte-level mutational fuzzing of serialized messages. + +Ported from jelly-jvm's `ByteFuzzer`. Valid messages are edited at random. Some +of them still parse as protobuf, but their contents are invalid, and the reader +must reject them with its own error. + +Runs are seeded, so a finding can be reproduced from the seed and iteration. Set +the JELLY_FUZZ_ITERATIONS and JELLY_FUZZ_SEED environment variables to change +them, e.g. for a longer run. +""" + +from __future__ import annotations + +import os +import random +import traceback +from collections.abc import Callable, Sequence +from typing import Final + +ITERATIONS: Final = int(os.environ.get("JELLY_FUZZ_ITERATIONS", "30000")) +SEED: Final = int(os.environ.get("JELLY_FUZZ_SEED", "20260918")) + +MAX_REPORTED: Final = 5 +MAX_SIZE: Final = 1 << 16 + +# Bytes with a special meaning in protobuf: the smallest and largest one-byte +# varints, a continuation bit, and the tag of field 1 +INTERESTING: Final = (0x00, 0x01, 0x7F, 0x80, 0xFF, 0x0A) + + +def mutate(data: bytes, rnd: random.Random) -> bytes: + """Apply 1-4 random edits to the bytes.""" + for _ in range(rnd.randrange(1, 5)): + edited = _edit(data, rnd) + if len(edited) <= MAX_SIZE: + data = edited + return data + + +def _edit(data: bytes, rnd: random.Random) -> bytes: # noqa: PLR0911 + if not data: + return data + size = len(data) + at = rnd.randrange(size) + edit = rnd.randrange(6) + if edit == 0: + # Flip one bit + return data[:at] + bytes([data[at] ^ (1 << rnd.randrange(8))]) + data[at + 1 :] + if edit == 1: + # Replace one byte with a random value + return data[:at] + bytes([rnd.randrange(256)]) + data[at + 1 :] + if edit == 2: # noqa: PLR2004 + # Replace one byte with one of INTERESTING + return data[:at] + bytes([rnd.choice(INTERESTING)]) + data[at + 1 :] + if edit == 3: # noqa: PLR2004 + # Cut the message short + return data[:at] + if edit == 4: # noqa: PLR2004 + # Repeat a slice, so repeated fields and submessages get longer + end = at + rnd.randrange(size - at) + 1 + return data[:end] + data[at:end] + data[end:] + # Insert 1-4 random bytes + return data[:at] + rnd.randbytes(rnd.randrange(1, 5)) + data[at:] + + +def findings( + corpus: Sequence[bytes], + consume: Callable[[bytes], object], + expected: tuple[type[BaseException], ...], + iterations: int = ITERATIONS, + seed: int = SEED, +) -> list[str]: + """ + Feed mutated copies of the corpus to `consume`. + + Returns: + list[str]: one description per exception that is not `expected`, with + enough to reproduce it. + + """ + rnd = random.Random(seed) # noqa: S311 + found: list[str] = [] + count = 0 + for i in range(iterations): + data = mutate(rnd.choice(corpus), rnd) + try: + consume(data) + except expected: + pass + except Exception as error: # noqa: BLE001 + count += 1 + if len(found) < MAX_REPORTED: + found.append(_describe(seed, i, data, error)) + if count > len(found): + found.append(f"...and {count - len(found)} more") + return found + + +def _describe(seed: int, iteration: int, data: bytes, error: Exception) -> str: + where = traceback.extract_tb(error.__traceback__)[-1] + more = f"... ({len(data)} bytes)" if len(data) > 96 else "" # noqa: PLR2004 + return ( + f"seed {seed}, iteration {iteration}: {type(error).__name__}: {error}\n" + f" at {where.filename}:{where.lineno}\n" + f" input: {data[:96].hex()}{more}" + )