Python: Detect invalid magic when instantiating SeekingReader

python/mcap/mcap/reader.py

@@ -34,7 +34,7 @@
     Schema,
     Statistics,
 )
-from .stream_reader import MAGIC_SIZE, StreamReader, breakup_chunk
+from .stream_reader import MAGIC_SIZE, StreamReader, breakup_chunk, read_magic
 from .summary import Summary
 
 
@@ -249,6 +249,7 @@ def __init__(
         decoder_factories: Iterable[DecoderFactory] = (),
     ):
         super().__init__(decoder_factories=decoder_factories)
+        read_magic(ReadDataStream(stream, calculate_crc=False))
         self._stream = stream
         self._validate_crcs = validate_crcs
         self._summary: Optional[Summary] = None

python/mcap/tests/test_reader.py

@@ -7,7 +7,7 @@
 import pytest
 
 from mcap.decoder import DecoderFactory
-from mcap.exceptions import DecoderNotFoundError
+from mcap.exceptions import DecoderNotFoundError, InvalidMagic
 from mcap.reader import McapReader, NonSeekingReader, SeekingReader, make_reader
 from mcap.records import Channel, Message, Schema
 from mcap.writer import IndexType, Writer
@@ -238,3 +238,17 @@ def test_no_summary_not_seeking(tmpdir: Path):
         assert len(list(NonSeekingReader(f).iter_attachments())) == 1
     with open(filepath, "rb") as f:
         assert len(list(NonSeekingReader(f).iter_metadata())) == 1
+
+
+def test_detect_invalid_initial_magic(tmpdir: Path):
+    filepath = tmpdir / "invalid_magic.mcap"
+    with open(filepath, "w") as f:
+        f.write("some bytes longer than the initial magic bytes")
+
+    with open(filepath, "rb") as f:
+        with pytest.raises(InvalidMagic):
+            SeekingReader(f)
+
+    with open(filepath, "rb") as f:
+        with pytest.raises(InvalidMagic):
+            NonSeekingReader(f).get_header()