Reject reports dated in the future and bound what parse_report_file() reads (#954)

* Reject reports whose index-naming date is more than two days in the future (#947)

The date that names a report's dated Elasticsearch/OpenSearch index (an
aggregate report's begin date, a failure report's Arrival-Date, an SMTP
TLS report's start-datetime) was never compared with the present, so a
report dated 2100 named an index for 2100. Each parser now rejects such a
report with its own Invalid*Report; past dates stay unbounded.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

* Refuse an input larger than the report size cap before parse_report_file() reads it (#949)

The 32 MiB cap was applied only after the whole input was read. A path is
now checked with os.path.getsize(), bytes-like input with len(), and a
file object is read at most one byte past the cap. The error is the
ParserError with the message the content check already gives. Three
existing tests that fed over-cap content now see the earlier rejection.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

* Measure a memoryview input by its byte size

len() of a memoryview counts items, which are not bytes for every format,
so the size cap in parse_report_file() compares nbytes instead.

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>

* Bound the aggregate end date and read file objects to their end

Address the Copilot review of #954. A normalized report's per-day records
name their indexes from their own interval starts, so the report's end date
is bounded like its begin date. A raw stream may return fewer bytes than
asked for before its end, so parse_report_file() now reads in a loop until
end of file or one byte past the cap.

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>

* Accumulate bounded reads in one buffer

Address the second Copilot review of #954. Keeping every chunk in a list
cost one bytes object per read, about 85 bytes per byte for a stream that
returns a byte at a time (22.7 MiB for 256 KiB of input). One bytearray
keeps memory proportional to the bytes read.

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Changed
 
+- **`parse_report_file()` now refuses an input larger than 32 MiB before reading it into memory** ([#949](https://github.com/domainaware/parsedmarc/issues/949)). The size cap (`MAX_DECOMPRESSED_REPORT_SIZE`) was applied only after the whole input had been read, so a path, a `bytes` object or a file object larger than the cap was held in memory first. The size of a path and the length of a `bytes`, `bytearray` or `memoryview` input are now checked up front, and a file object is read at most one byte past the cap (it is left positioned there). The input is rejected with `ParserError` and the same message as before (`the report is larger than 32 MiB`). A valid compressed report is smaller than its decompressed form, so the one cap applies to both.
+- **A report dated more than two days in the future is now rejected** ([#947](https://github.com/domainaware/parsedmarc/issues/947)). With Elasticsearch or OpenSearch output a report's own date names its dated index (`dmarc_aggregate-YYYY-MM-DD`), and nothing compared that date with the present, so a report with `<begin>4102444800</begin>` created an index for the year 2100. An aggregate report whose begin or end date (a report split per day names an index from each day it covers), an SMTP TLS report whose `start-datetime`, or a failure report whose `Arrival-Date` is more than 2 days ahead of the current time (`MAX_REPORT_FUTURE_DAYS`) is now rejected with `InvalidAggregateReport`, `InvalidSMTPTLSReport` or `InvalidFailureReport`, so a mailbox message is moved to the `Invalid` folder. The two days absorb a reporter's clock skew and time-zone mistakes; dates in the past are not bounded, so archives of old reports still parse.
 - **Python 3.10 is no longer supported, and Python 3.15 is.** Python 3.10 reached end of life in October 2026; `requires-python` is now `>=3.11`. Python 3.15 is now tested in CI and listed as supported.
 
 ### Bug fixes

docs/source/elasticsearch.md

@@ -420,6 +420,10 @@ index with their own dates, and a warning is logged. Dated index names still com
 from each report's own dates, so reports with many different dates
 still create one index per date; data streams (`use_data_streams`) are
 affected by neither.
+A report whose begin or end date (aggregate), begin date (SMTP TLS) or
+`Arrival-Date` (failure) is more than two days in the future is rejected
+as invalid, because those dates name the dated indexes; past dates are
+not bounded.
 
 Set `use_data_streams = True` in the `[elasticsearch]` section to write
 to [data streams] instead. Each report type then goes to one data stream

docs/source/opensearch.md

@@ -21,6 +21,10 @@ index with their own dates, and a warning is logged. Dated index names still com
 from each report's own dates, so reports with many different dates
 still create one index per date; data streams (`use_data_streams`) are
 affected by neither.
+A report whose begin or end date (aggregate), begin date (SMTP TLS) or
+`Arrival-Date` (failure) is more than two days in the future is rejected
+as invalid, because those dates name the dated indexes; past dates are
+not bounded.
 
 Set `use_data_streams = True` in the `[opensearch]` section to write to
 [data streams] instead. Each report type then goes to one data stream

parsedmarc/__init__.py

@@ -53,6 +53,7 @@ from parsedmarc.constants import (
     DEFAULT_DNS_MAX_RETRIES,
     DEFAULT_DNS_TIMEOUT,
     MAX_NORMALIZED_RECORDS,
+    MAX_REPORT_FUTURE_DAYS,
     MAX_REPORT_SPAN_DAYS,
     MAX_XML_DEPTH,
     MAX_XML_NODES,
@@ -1040,7 +1041,10 @@ def _smtp_tls_datetime(value: Any, field: str) -> str:
     Splunk output (a UNIX timestamp) make later. dateutil also accepts a
     partial date such as ``"2016"``, which PostgreSQL may still reject; that
     failure is raised as ``PostgreSQLError`` and recorded as an output
-    error. Raises ``InvalidSMTPTLSReport`` otherwise.
+    error. A ``start-datetime`` more than ``MAX_REPORT_FUTURE_DAYS`` days in
+    the future is also rejected, because it names the dated index (#947);
+    ``end-datetime`` is not checked. Raises ``InvalidSMTPTLSReport``
+    otherwise.
     """
     value = _smtp_tls_string(value, field)
     if len(value) > _SMTP_TLS_MAX_DATETIME_LENGTH:
@@ -1048,12 +1052,17 @@ def _smtp_tls_datetime(value: Any, field: str) -> str:
             f"{field} is longer than {_SMTP_TLS_MAX_DATETIME_LENGTH} characters"
         )
     try:
-        human_timestamp_to_datetime(value, to_utc=True)
+        moment = human_timestamp_to_datetime(value, to_utc=True)
         human_timestamp_to_unix_timestamp(value)
     except (ValueError, OverflowError) as e:
         raise InvalidSMTPTLSReport(
             f"{field} is not a valid date-time: {sanitize_for_log(value)}"
         ) from e
+    if field == "start-datetime":
+        try:
+            _check_not_future(moment, field)
+        except ValueError as error:
+            raise InvalidSMTPTLSReport(str(error)) from error
     return value
 
 
@@ -1516,6 +1525,19 @@ def _check_report_span(span_seconds: int) -> None:
         )
 
 
+def _check_not_future(moment: datetime, what: str) -> None:
+    """Raise ``ValueError`` when ``moment`` is more than
+    ``MAX_REPORT_FUTURE_DAYS`` days ahead of the current time, because the
+    date that names a report's dated index would then create an index for a
+    date far in the future (#947). ``moment`` must be an aware UTC datetime;
+    each parser converts the error to its own ``Invalid*Report``."""
+    if moment > datetime.now(timezone.utc) + timedelta(days=MAX_REPORT_FUTURE_DAYS):
+        raise ValueError(
+            f"{what} {moment.isoformat()} is more than "
+            f"{MAX_REPORT_FUTURE_DAYS} days in the future"
+        )
+
+
 def _days_touched(begin: datetime, end: datetime) -> int:
     """Return how many calendar days of ``begin``'s timezone the interval
     [begin, end) touches: the number of buckets ``_bucket_interval_by_day()``
@@ -1747,6 +1769,17 @@ def parse_aggregate_report_xml(
         end_ts = int(date_range["end"].split(".")[0])
         span_seconds = end_ts - begin_ts
         _check_report_span(span_seconds)
+        # The end is checked as well as the begin: a normalized report's
+        # per-day records each name an index from their own interval start,
+        # and every interval start lies at or before the report's end.
+        try:
+            for stamp, what in (
+                (begin_ts, "Report begin date"),
+                (end_ts, "Report end date"),
+            ):
+                _check_not_future(datetime.fromtimestamp(stamp, timezone.utc), what)
+        except ValueError as error:
+            raise InvalidAggregateReport(str(error)) from error
 
         normalize_timespan = (
             span_seconds > cfg.normalize_timespan_threshold_hours * 3600
@@ -2432,6 +2465,10 @@ def parse_failure_report(
             raise InvalidFailureReport(
                 f"Invalid Arrival-Date: {sanitize_for_log(str(error))}"
             ) from error
+        try:
+            _check_not_future(arrival_utc, "Arrival-Date")
+        except ValueError as error:
+            raise InvalidFailureReport(str(error)) from error
         arrival_utc = arrival_utc.strftime("%Y-%m-%d %H:%M:%S")
         parsed_report["arrival_date_utc"] = arrival_utc
 
@@ -3089,6 +3126,22 @@ def _describe_parse_failure(
     )
 
 
+def _read_up_to(file_object: BinaryIO, limit: int) -> bytes:
+    """Read from ``file_object`` until end of file or ``limit`` bytes,
+    whichever comes first. A raw or unbuffered stream (a socket file, a
+    pipe) may return fewer bytes than asked for without being at its end, so
+    one sized ``read()`` could hand the parser a truncated report."""
+    # One growing buffer, not a list of chunks: a stream that returns a byte
+    # at a time would otherwise cost one bytes object per byte read.
+    content = bytearray()
+    while len(content) < limit:
+        chunk = file_object.read(limit - len(content))
+        if not chunk:
+            break
+        content += chunk
+    return bytes(content)
+
+
 def parse_report_file(
     input_: bytes | str | os.PathLike[str] | os.PathLike[bytes] | BinaryIO,
     *,
@@ -3112,6 +3165,8 @@ def parse_report_file(
     object is read from but never closed, on either the success or the
     exception path; it is left open and positioned wherever its own
     ``read()`` left it, so the caller is free to ``seek(0)`` and reuse it.
+    When the input is over the size cap, that is one byte past the cap: no
+    more is read.
 
     Args:
         input_ (str | os.PathLike | bytes | BinaryIO): A path to a file,
@@ -3138,6 +3193,11 @@ def parse_report_file(
 
     Returns:
         dict: The parsed DMARC report
+
+    Raises:
+        ParserError: if the input is not a valid report, or is larger than
+            ``MAX_DECOMPRESSED_REPORT_SIZE`` (refused before it is read into
+            memory)
     """
     cfg = _resolve_config(
         config,
@@ -3153,15 +3213,23 @@ def parse_report_file(
         normalize_timespan_threshold_hours=normalize_timespan_threshold_hours,
     )
     file_object: BinaryIO
+    # An input over the cap is refused before it is held in memory (#949). A
+    # valid compressed report is smaller than its decompressed form, so the
+    # one cap serves both; the message is the one the content check gives.
+    too_large = ParserError(_size_cap_text(b""))
     if isinstance(input_, (str, os.PathLike)):
         file_path = os.fspath(input_)
         logger.debug(f"Parsing {file_path}")
+        if os.path.getsize(file_path) > MAX_DECOMPRESSED_REPORT_SIZE:
+            raise too_large
         # A path is a handle we opened ourselves, so it must be closed on
         # both the success and exception paths.
         with open(file_path, "rb") as file_object:
-            content = file_object.read()
+            content = _read_up_to(file_object, MAX_DECOMPRESSED_REPORT_SIZE + 1)
     else:
         if isinstance(input_, (bytes, bytearray, memoryview)):
+            if memoryview(input_).nbytes > MAX_DECOMPRESSED_REPORT_SIZE:
+                raise too_large
             # The BytesIO wrapper is created here, so this function owns
             # it and closes it once the bytes have been read out.
             file_object = BytesIO(bytes(input_))
@@ -3173,7 +3241,9 @@ def parse_report_file(
             # seek(0) and retry, inspect tell(), or otherwise reuse the
             # handle afterward.
             file_object = input_
-            content = file_object.read()
+            content = _read_up_to(file_object, MAX_DECOMPRESSED_REPORT_SIZE + 1)
+    if len(content) > MAX_DECOMPRESSED_REPORT_SIZE:
+        raise too_large
     if content.startswith(MAGIC_ZIP) or content.startswith(MAGIC_GZIP):
         content = extract_report(content)
 

parsedmarc/constants.py

@@ -36,6 +36,11 @@ MAX_XML_TAG_LENGTH = 65_536
 # interval is 86400 seconds); a year leaves room for mistakes, and
 # normalization emits at most one record per day.
 MAX_REPORT_SPAN_DAYS = 366
+# The date that names a report's dated index (Elasticsearch, OpenSearch) may
+# not be more than this many days ahead of the current time. Two days absorb
+# a reporter's clock skew and time-zone mistakes. The past is not bounded: an
+# archive of old reports is legitimate input.
+MAX_REPORT_FUTURE_DAYS = 2
 # Most distinct dated indexes one report may create in Elasticsearch or
 # OpenSearch (one shard each); further records go to the report's first index.
 MAX_INDEXES_PER_REPORT = 31

tests/test_init.py

@@ -10,6 +10,7 @@ import binascii
 import csv
 import email
 import email.errors
+import email.utils
 import gc
 import gzip
 import inspect
@@ -18,6 +19,7 @@ import logging
 import mailbox
 import os
 import quopri
+import re
 import time
 import traceback
 import tracemalloc
@@ -31,6 +33,7 @@ from email.mime.multipart import MIMEMultipart
 from email.mime.nonmultipart import MIMENonMultipart
 from email.mime.text import MIMEText
 from glob import glob
+import io
 from io import BytesIO, StringIO
 from pathlib import Path
 from shutil import rmtree
@@ -3220,13 +3223,15 @@ class TestParseReportFile(unittest.TestCase):
         finally:
             tracemalloc.stop()
         email_parser.assert_not_called()
-        self.assertIn("exceeds the 33554432 byte limit", str(ctx.exception))
+        # parse_report_file() refuses the input before parsing any of it.
+        self.assertEqual(str(ctx.exception), "the report is larger than 32 MiB")
         self.assertLess(elapsed, limit)
         self.assertLess(peak, 100 * 2**20)
 
     def testParseReportFileCapsTextForTheEmailParser(self):
         """Text over the size cap that does not look like XML or JSON is not
-        handed to the email parser either."""
+        handed to the email parser either (refused before it is read,
+        #949)."""
         content = b"From: a@example.com\n\n" + b"x" * (
             parsedmarc.MAX_DECOMPRESSED_REPORT_SIZE + 1
         )
@@ -3234,9 +3239,7 @@ class TestParseReportFile(unittest.TestCase):
             with self.assertRaises(parsedmarc.ParserError) as ctx:
                 parsedmarc.parse_report_file(content, offline=True)
         email_parser.assert_not_called()
-        self.assertEqual(
-            str(ctx.exception), "Invalid report email: the report is larger than 32 MiB"
-        )
+        self.assertEqual(str(ctx.exception), "the report is larger than 32 MiB")
 
     def testParseReportFileSkipsParsersThatCannotMatch(self):
         """Content that begins with "<" is not tried as JSON or as an email,
@@ -3366,7 +3369,7 @@ class TestParseReportFile(unittest.TestCase):
             def __init__(self):
                 self.closed = False
 
-            def read(self):
+            def read(self, size=-1):
                 raise OSError("boom")
 
             def close(self):
@@ -3381,7 +3384,11 @@ class TestParseReportFile(unittest.TestCase):
 
         fake_handle = _RaisingHandle()
 
-        with patch("builtins.open", return_value=fake_handle) as mock_open:
+        # The size is checked before the file is opened.
+        with (
+            patch("os.path.getsize", return_value=10),
+            patch("builtins.open", return_value=fake_handle) as mock_open,
+        ):
             with self.assertRaises(OSError):
                 parsedmarc.parse_report_file("some/path.xml", offline=True)
 
@@ -3409,7 +3416,7 @@ class TestParseReportFile(unittest.TestCase):
             def __init__(self):
                 self.close_called = False
 
-            def read(self):
+            def read(self, size=-1):
                 raise OSError("boom")
 
             def close(self):
@@ -9431,5 +9438,240 @@ class TestFailureReportPartPrecedence(unittest.TestCase):
         self._assert_arf(self._parse(self.FEEDBACK + sample))
 
 
+class TestReportDateInTheFuture(unittest.TestCase):
+    """A report whose index-naming date is more than MAX_REPORT_FUTURE_DAYS
+    ahead is rejected (#947): that date names the dated Elasticsearch or
+    OpenSearch index, and nothing bounded it against the present."""
+
+    AGGREGATE = "samples/aggregate/!example.com!1538204542!1538463818.xml"
+    SMTP_TLS = "samples/smtp_tls/rfc8460.json"
+    FAILURE = "samples/failure/dmarc_ruf_report_linkedin.crlf.eml"
+
+    @staticmethod
+    def _tomorrow() -> datetime:
+        return (datetime.now(timezone.utc) + timedelta(days=1)).replace(microsecond=0)
+
+    def _aggregate(self, moment: datetime | None) -> str:
+        with open(self.AGGREGATE, encoding="utf-8") as f:
+            xml = f.read()
+        if moment is None:
+            return xml
+        stamp = str(int(moment.timestamp()))
+        xml = re.sub(r"<begin>\d+</begin>", f"<begin>{stamp}</begin>", xml)
+        return re.sub(r"<end>\d+</end>", f"<end>{stamp}</end>", xml)
+
+    def _parse_failure(self, arrival: str):
+        feedback = "\n".join(
+            [
+                "Feedback-Type: auth-failure",
+                "User-Agent: test/1.0",
+                "Version: 1",
+                "Original-Mail-From: sender@example.com",
+                f"Arrival-Date: {arrival}",
+                "Source-IP: 203.0.113.1",
+                "Reported-Domain: example.com",
+                "Auth-Failure: dmarc",
+            ]
+        )
+        return parsedmarc.parse_failure_report(
+            feedback,
+            "From: a@example.com\n\nbody",
+            datetime(2024, 1, 1, tzinfo=timezone.utc),
+            offline=True,
+        )
+
+    def _tls(self, start: str, end: str) -> str:
+        with open(self.SMTP_TLS, encoding="utf-8") as f:
+            report = json.load(f)
+        report["date-range"] = {"start-datetime": start, "end-datetime": end}
+        return json.dumps(report)
+
+    def testAggregateInTheFutureIsRejected(self):
+        xml = self._aggregate(datetime(2100, 1, 1, tzinfo=timezone.utc))
+        with self.assertRaises(parsedmarc.InvalidAggregateReport) as ctx:
+            parsedmarc.parse_aggregate_report_xml(xml, offline=True)
+        self.assertIn("2100-01-01", str(ctx.exception))
+        self.assertIn("in the future", str(ctx.exception))
+        self.assertNotIn("Unexpected error", str(ctx.exception))
+
+    def testAggregateTomorrowIsAccepted(self):
+        tomorrow = self._tomorrow()
+        report = parsedmarc.parse_aggregate_report_xml(
+            self._aggregate(tomorrow), offline=True
+        )
+        self.assertEqual(
+            report["report_metadata"]["begin_date"],
+            # begin_date is rendered in the machine's local time zone
+            datetime.fromtimestamp(tomorrow.timestamp()).strftime("%Y-%m-%d %H:%M:%S"),
+        )
+
+    def testAggregateEndingInTheFutureIsRejected(self):
+        """A normalized report's per-day records name their indexes from
+        their own interval starts, so a report that begins now and ends in
+        30 days would create indexes up to a month ahead. The end date is
+        bounded like the begin date."""
+        now = datetime.now(timezone.utc)
+        xml = self._aggregate(now)
+        end = str(int((now + timedelta(days=30)).timestamp()))
+        xml = re.sub(r"<end>\d+</end>", f"<end>{end}</end>", xml)
+        with self.assertRaises(parsedmarc.InvalidAggregateReport) as ctx:
+            parsedmarc.parse_aggregate_report_xml(xml, offline=True)
+        self.assertIn("Report end date", str(ctx.exception))
+        self.assertIn("in the future", str(ctx.exception))
+
+    def testAggregateSampleStillParses(self):
+        report = parsedmarc.parse_aggregate_report_xml(
+            self._aggregate(None), offline=True
+        )
+        self.assertEqual(report["report_metadata"]["begin_date"][:4], "2018")
+
+    def testFailureInTheFutureIsRejected(self):
+        with self.assertRaises(parsedmarc.InvalidFailureReport) as ctx:
+            self._parse_failure("Fri, 01 Jan 2100 00:00:00 +0000")
+        self.assertIn("2100-01-01", str(ctx.exception))
+        self.assertIn("in the future", str(ctx.exception))
+        self.assertNotIn("Unexpected error", str(ctx.exception))
+
+    def testFailureTomorrowIsAccepted(self):
+        tomorrow = self._tomorrow()
+        report = self._parse_failure(email.utils.format_datetime(tomorrow))
+        self.assertEqual(
+            report["arrival_date_utc"], tomorrow.strftime("%Y-%m-%d %H:%M:%S")
+        )
+
+    def testFailureSampleStillParses(self):
+        result = parsedmarc.parse_report_file(self.FAILURE, offline=True)
+        self.assertEqual(result["report_type"], "failure")
+
+    def testSmtpTlsInTheFutureIsRejected(self):
+        report = self._tls("2100-01-01T00:00:00Z", "2100-01-01T23:59:59Z")
+        with self.assertRaises(parsedmarc.InvalidSMTPTLSReport) as ctx:
+            parsedmarc.parse_smtp_tls_report_json(report)
+        self.assertIn("start-datetime", str(ctx.exception))
+        self.assertIn("2100-01-01", str(ctx.exception))
+        self.assertIn("in the future", str(ctx.exception))
+
+    def testSmtpTlsTomorrowIsAccepted(self):
+        tomorrow = self._tomorrow()
+        report = parsedmarc.parse_smtp_tls_report_json(
+            self._tls(tomorrow.isoformat(), tomorrow.isoformat())
+        )
+        self.assertEqual(report["begin_date"], tomorrow.isoformat())
+
+    def testSmtpTlsSampleStillParses(self):
+        with open(self.SMTP_TLS, encoding="utf-8") as f:
+            report = parsedmarc.parse_smtp_tls_report_json(f.read())
+        self.assertEqual(report["begin_date"], "2016-04-01T00:00:00Z")
+
+    def testSmtpTlsEndDatetimeInTheFutureIsAccepted(self):
+        """Only start-datetime names the index, so a far-future
+        end-datetime is not rejected by this check (a TLS report's span is
+        not bounded)."""
+        report = parsedmarc.parse_smtp_tls_report_json(
+            self._tls("2016-04-01T00:00:00Z", "2100-01-01T00:00:00Z")
+        )
+        self.assertEqual(report["end_date"], "2100-01-01T00:00:00Z")
+
+
+class TestParseReportFileSizeCap(unittest.TestCase):
+    """parse_report_file() refuses an input over the size cap before holding
+    it in memory (#949), with the message the content check gives."""
+
+    CAP = parsedmarc.MAX_DECOMPRESSED_REPORT_SIZE
+    MESSAGE = "the report is larger than 32 MiB"
+
+    def testPathOverTheCapIsRejectedWithoutOpeningIt(self):
+        with NamedTemporaryFile() as f:
+            f.truncate(self.CAP + 1)
+            with patch("parsedmarc.open", create=True) as opened:
+                with self.assertRaises(parsedmarc.ParserError) as ctx:
+                    parsedmarc.parse_report_file(f.name, offline=True)
+            opened.assert_not_called()
+        self.assertEqual(str(ctx.exception), self.MESSAGE)
+
+    def testBytesOverTheCapAreRejected(self):
+        with self.assertRaises(parsedmarc.ParserError) as ctx:
+            # bytearray is accepted at run time, though annotated as bytes
+            parsedmarc.parse_report_file(
+                cast(bytes, bytearray(self.CAP + 1)), offline=True
+            )
+        self.assertEqual(str(ctx.exception), self.MESSAGE)
+
+    def testFileObjectOverTheCapStopsReading(self):
+        stream = BytesIO(bytes(self.CAP + 100))
+        with self.assertRaises(parsedmarc.ParserError) as ctx:
+            parsedmarc.parse_report_file(stream, offline=True)
+        self.assertEqual(str(ctx.exception), self.MESSAGE)
+        self.assertEqual(stream.tell(), self.CAP + 1)
+
+    def testShortReadingStreamIsReadToItsEnd(self):
+        """A raw stream may return fewer bytes than asked for before its
+        end. The whole report must still be read: a single sized read would
+        parse a truncated report."""
+
+        class HundredBytesAtATime(io.RawIOBase):
+            def __init__(self, data: bytes) -> None:
+                self._inner = io.BytesIO(data)
+
+            def readable(self) -> bool:
+                return True
+
+            def readinto(self, buffer) -> int:
+                chunk = self._inner.read(min(len(buffer), 100))
+                buffer[: len(chunk)] = chunk
+                return len(chunk)
+
+        path = "samples/aggregate/!example.com!1538204542!1538463818.xml"
+        with open(path, "rb") as f:
+            data = f.read()
+        self.assertGreater(len(data), 100)
+        result = parsedmarc.parse_report_file(
+            cast(BinaryIO, HundredBytesAtATime(data)), offline=True
+        )
+        self.assertEqual(result["report_type"], "aggregate")
+
+    def testByteAtATimeStreamDoesNotMultiplyMemory(self):
+        """Reading a stream that returns one byte per read costs memory in
+        proportion to the bytes read, not to the number of reads. Keeping
+        every chunk in a list cost about 85 bytes per byte read (89 MiB for
+        1 MiB of input), so 256 KiB read that way peaked above 20 MiB; one
+        growing buffer stays under 2 MiB."""
+
+        class OneByteAtATime(io.RawIOBase):
+            def __init__(self, data: bytes) -> None:
+                self._inner = io.BytesIO(data)
+
+            def readable(self) -> bool:
+                return True
+
+            def readinto(self, buffer) -> int:
+                chunk = self._inner.read(1)
+                buffer[: len(chunk)] = chunk
+                return len(chunk)
+
+        size = 256 * 1024
+        stream = cast(BinaryIO, OneByteAtATime(b"x" * size))
+        tracemalloc.start()
+        try:
+            content = parsedmarc._read_up_to(stream, size + 1)
+            peak = tracemalloc.get_traced_memory()[1]
+        finally:
+            tracemalloc.stop()
+        self.assertEqual(len(content), size)
+        self.assertLess(peak, 2 * 1024 * 1024)
+
+    def testInputOfExactlyTheCapIsNotRejectedForSize(self):
+        """Zero bytes are not a report, but the size check does not fire at
+        the cap."""
+        with NamedTemporaryFile() as f:
+            f.truncate(self.CAP)
+            inputs = [f.name, bytearray(self.CAP), BytesIO(bytes(self.CAP))]
+            for input_ in inputs:
+                with self.subTest(type=type(input_).__name__):
+                    with self.assertRaises(parsedmarc.ParserError) as ctx:
+                        parsedmarc.parse_report_file(input_, offline=True)
+                    self.assertNotEqual(str(ctx.exception), self.MESSAGE)
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)