Add hardening tests

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)
 

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,))

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:

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

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))