diff --git a/README.md b/README.md index 1a60a48..2ab8679 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,47 @@ is left to the caller. Version 3 is the current version produced for new OWIDs. Versions 1 and 2 are deprecated and are supported for reading existing data only. +## Payload size and application limits + +The OWID wire format stores the payload length as an unsigned 32 bit value, +so a payload from zero through 4,294,967,295 bytes is structurally valid. The +format defines no smaller payload limit. The null-terminated domain carries +no length of its own, so the protocol alone is not an application input limit +for the complete envelope. + +This package validates that the declared payload length agrees with the bytes +present before it sizes or copies the payload. A large declaration without +the corresponding bytes is malformed and is rejected without allocating the +declared size. A matching large payload is not malformed merely because it is +large, and parsing work and memory use scale with the bytes actually present. + +The domain ends at a zero terminator rather than at a declared length, so a +buffer whose terminator is missing or corrupted would otherwise be walked to +its end. This package stops that walk at `MAXIMUM_DOMAIN_LENGTH`, which +`owid/io.py` derives from the size limit in RFC 1035 section 2.3.4, and +refuses the buffer there. The cost of a domain field an attacker sized is +therefore fixed by that constant rather than by the length of the input, and +no domain a name server would accept is affected. + +The same maximum binds the write, because a library that emits something it +cannot read moves the fault to the consumer. A `Creator` refuses a domain +longer than `MAXIMUM_DOMAIN_LENGTH` when the caller supplies it, before any +signing work is done, and the writer refuses one that reached an OWID by any +other route when the OWID is serialised. Both raise `OwidError` naming the +maximum, as the parse does. + +The in-memory APIs remain subject to Python object, address-space and +available-memory limits. Applications accepting untrusted OWIDs must choose +limits suitable for their use case and enforce them before buffering the +binary form or decoding Base64. An implementation capacity failure or an +application policy rejection is distinct from an invalid OWID. + +For transport input, limit the complete HTTP body or encoded envelope; allow +for the domain and other OWID fields as well as the payload. After parsing, +`len(owid.payload)` reports the actual payload size without another copy and +can be used for downstream policy. The parser cannot choose either limit on +behalf of the application. + ## Installation The package targets Python 3.9 and later and depends on the diff --git a/owid/creator.py b/owid/creator.py index 0330376..bc0346a 100644 --- a/owid/creator.py +++ b/owid/creator.py @@ -27,7 +27,7 @@ from .crypto import Crypto from .error import OwidError -from .io import SIGNATURE_LENGTH +from .io import MAXIMUM_DOMAIN_LENGTH, SIGNATURE_LENGTH from .owid import Owid from .version import DEFAULT_VERSION @@ -62,11 +62,22 @@ def __init__(self, domain: str, crypto: Crypto) -> None: """Creates a new creator for the domain using the crypto instance for signing. - Raises OwidError if the domain is empty or whitespace, or the crypto - instance can not sign. + Raises OwidError if the domain is empty or whitespace, longer than + the published maximum, or the crypto instance can not sign. + + The domain is checked here, where the caller supplies it, rather than + only when an OWID is written, so a creator that could only produce + OWIDs this library refuses to read never exists and no signing work + is done before the refusal. """ if domain is None or domain.strip() == "": raise OwidError("domain '{0}' is not valid".format(domain)) + if len(domain.encode("utf-8")) > MAXIMUM_DOMAIN_LENGTH: + raise OwidError( + "domain is longer than the '{0}' character maximum".format( + MAXIMUM_DOMAIN_LENGTH + ) + ) if not crypto.can_sign(): raise OwidError("instance of Crypto cannot be used to generate a signature") self._domain = domain @@ -77,8 +88,9 @@ def from_configuration(cls, configuration: Configuration) -> "Creator": """Creates a new creator from configuration containing the domain and the private key PEM. - Raises OwidError if the domain is empty or whitespace, or the private - key PEM is not valid. + Raises OwidError if the domain is empty or whitespace, the domain is + longer than the published maximum, or the private key PEM is not + valid. """ crypto = Crypto.new_sign_only(configuration.private_key) return cls(configuration.domain, crypto) diff --git a/owid/io.py b/owid/io.py index 6667291..da35a26 100644 --- a/owid/io.py +++ b/owid/io.py @@ -36,6 +36,15 @@ #: number of hours or minutes after this instant. BASE_DATE = datetime(2020, 1, 1, tzinfo=timezone.utc) +#: The longest creator domain the reader will accept, in characters. RFC 1035 +#: section 2.3.4, "Size limits", restricts the total length of a domain name, +#: being the label octets and the label length octets, to 255 octets or less. +#: An OWID stores the presentation form, the text "example.com", where the +#: dots stand in for the label length octets and the root label has no text at +#: all, so two of those 255 octets have no text equivalent and the limit on +#: the text is two fewer. A domain is ASCII, so a character is a byte here. +MAXIMUM_DOMAIN_LENGTH = 255 - 2 + class Reader: """Sequential reader over a byte buffer.""" @@ -62,26 +71,61 @@ def read_bytes(self, count: int) -> bytes: return value def read_string(self) -> str: - """Reads bytes up to the null terminator and decodes them as UTF-8.""" - remaining = self._buffer[self._position:] - terminator = remaining.find(0) + """Reads bytes up to the null terminator and decodes them as UTF-8. + + The only null terminated string in an OWID is the creator domain, and + a domain has a published maximum length, so the search for the + terminator stops after MAXIMUM_DOMAIN_LENGTH bytes rather than running + to the end of the buffer. A buffer whose terminator is missing or + corrupted is refused as soon as that window is exhausted, so the work + a hostile buffer can ask for is fixed by the constant rather than + growing with the length of the input. + """ + window_end = self._position + MAXIMUM_DOMAIN_LENGTH + 1 + terminator = self._buffer.find(b"\0", self._position, window_end) if terminator < 0: - raise OwidError("buffer ended before the OWID was complete") + if len(self._buffer) < window_end: + raise OwidError("buffer ended before the OWID was complete") + raise OwidError( + "domain is longer than the '{0}' character maximum".format( + MAXIMUM_DOMAIN_LENGTH + ) + ) try: - value = remaining[:terminator].decode("utf-8") + value = self._buffer[self._position:terminator].decode("utf-8") except UnicodeDecodeError: raise OwidError("domain bytes are not valid UTF-8") - self._position += terminator + 1 + self._position = terminator + 1 return value def read_u32(self) -> int: """Reads an unsigned 32 bit little endian integer.""" - return struct.unpack(" len(self._buffer): + raise OwidError("buffer ended before the OWID was complete") + value = struct.unpack_from(" bytes: - """Reads a byte array prefixed with its length as an unsigned 32 bit - integer.""" + """Reads the payload, being a byte array prefixed with its length as + an unsigned 32 bit integer. + + The length is whatever the sender declared, so it is checked against + the bytes actually present before anything is sized by it. A valid + OWID is the declared payload followed by the signature and nothing + else, so the length must equal the bytes remaining less the signature + length, and any other length, short or long, is refused here. The + same check refuses a signature shorter than 64 bytes and any byte + after the signature, which until 28 August 2026 this reader ignored. + """ count = self.read_u32() + remaining = len(self._buffer) - self._position + if remaining != count + SIGNATURE_LENGTH: + raise OwidError( + "OWID payload length '{0}' does not match the '{1}' bytes " + "present, of which the final '{2}' must be the " + "signature".format(count, remaining, SIGNATURE_LENGTH) + ) return self.read_bytes(count) def read_signature(self) -> bytes: @@ -111,10 +155,23 @@ def write_string(buffer: bytearray, value: str) -> None: The string must not contain a null character because that would conflict with the terminator. + + The only string written this way is the creator domain, and the reader + refuses a domain longer than MAXIMUM_DOMAIN_LENGTH, so a longer one is + refused here as well and the library never emits an OWID that it would + then refuse to read. The length compared is the encoded bytes, because + those are what the reader walks, and for the ASCII a domain is made of + they are the same count as the characters. """ encoded = value.encode("utf-8") if 0 in encoded: raise OwidError("domain '{0}' is not valid".format(value)) + if len(encoded) > MAXIMUM_DOMAIN_LENGTH: + raise OwidError( + "domain is longer than the '{0}' character maximum".format( + MAXIMUM_DOMAIN_LENGTH + ) + ) buffer.extend(encoded) buffer.append(0) diff --git a/owid/owid.py b/owid/owid.py index 0ed54ed..486e839 100644 --- a/owid/owid.py +++ b/owid/owid.py @@ -94,8 +94,11 @@ def from_base64(cls, value: str) -> "Owid": def from_byte_array(cls, buffer: bytes) -> "Owid": """Creates an OWID from its binary form. - Raises OwidError if the first byte is not a known version or the - buffer is too short for the remaining fields. + The buffer must hold exactly one OWID, ending with the 64 byte + signature. Raises OwidError if the first byte is not a known + version, the buffer is too short for the remaining fields, or the + declared payload length does not leave exactly the signature at the + end of the buffer. """ reader = io.Reader(bytes(buffer)) return cls._from_reader(reader) diff --git a/tests/test_domain_length.py b/tests/test_domain_length.py new file mode 100644 index 0000000..9b5ea57 --- /dev/null +++ b/tests/test_domain_length.py @@ -0,0 +1,293 @@ +# **************************************************************************** +# Copyright 2026 51 Degrees Mobile Experts Limited (51degrees.com) +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# **************************************************************************** +"""Tests for the domain length bound in the OWID parse and in the write. + +The creator domain is stored as text followed by a zero terminator, so the +parse finds the end of the domain by walking forward to that terminator. If +the terminator is missing or corrupted the walk would otherwise run to the +end of the buffer, which is work an attacker chooses the size of. A domain +has a published maximum length, so these tests prove that a domain of the +maximum length still parses, that a longer one is refused, and that a buffer +whose terminator is missing or far away costs no more than the bound. + +The same maximum binds the write. A creator configured with a longer domain +would otherwise produce an OWID that this same library refuses to parse, so +the tests below prove that a longer domain is refused when the caller +supplies it to a creator and again when a domain that arrived by any other +route is serialised. +""" + +from __future__ import annotations + +import struct +import time +import tracemalloc +import unittest + +from owid import SIGNATURE_LENGTH, Creator, Crypto, Owid, OwidError, Version +from owid.io import MAXIMUM_DOMAIN_LENGTH, write_string + +#: A domain of exactly the maximum length, built from labels no longer than +#: the 63 characters RFC 1035 allows so that the value is a shape a real +#: domain could take rather than one long run of letters. +MAXIMUM_DOMAIN = ".".join(["a" * 63] * 3 + ["b" * 61]) + +PAYLOAD = bytes([0x5A]) * 37 +SIGNATURE = bytes([0x99]) * SIGNATURE_LENGTH + +#: The length of the domain field in the hostile buffers below. Large enough +#: that a walk over the whole of one is hundreds of thousands of times the +#: bounded walk, so the assertions on time and on memory separate the two +#: cases with room to spare. +HOSTILE_LENGTH = 64 * 1024 * 1024 + + +def hostile(domain_bytes: bytes) -> bytes: + """A buffer holding the version byte and the raw domain bytes given and + nothing after them. Both callers hand over a domain field the reader must + refuse, so nothing valid is needed after it, and the only zero byte in + either buffer is the one the caller puts there itself.""" + return bytes([Version.VERSION3.as_byte()]) + domain_bytes + + +def envelope(domain_bytes: bytes) -> bytes: + """A version 3 envelope built from the raw domain bytes given, being the + version byte, those bytes, four minute bytes, the payload with its length + and the signature. The domain bytes carry their own terminator, so a test + can set the length of the domain and leave the rest of the envelope + valid.""" + buffer = bytearray() + buffer.append(Version.VERSION3.as_byte()) + buffer.extend(domain_bytes) + buffer.extend(struct.pack(" bytes: + """The domain as ASCII followed by the zero terminator.""" + return domain.encode("ascii") + b"\0" + + +class DomainLengthTests(unittest.TestCase): + def test_maximum_length_domain_parses(self) -> None: + """A domain of exactly the maximum length parses and the value round + trips through the writer and the reader unchanged.""" + self.assertEqual(len(MAXIMUM_DOMAIN), MAXIMUM_DOMAIN_LENGTH) + + owid = Owid.from_byte_array(envelope(terminated(MAXIMUM_DOMAIN))) + + self.assertEqual(owid.domain, MAXIMUM_DOMAIN) + self.assertEqual(owid.payload, PAYLOAD) + self.assertEqual(owid.signature, SIGNATURE) + again = Owid.from_byte_array(owid.as_byte_array()) + self.assertEqual(again.domain, MAXIMUM_DOMAIN) + self.assertEqual(again, owid) + + def test_one_character_over_the_maximum_is_refused(self) -> None: + """A domain one character longer than the maximum is refused, and + the message names the maximum so the reader of a log can see which + limit the sender crossed.""" + domain = MAXIMUM_DOMAIN + "c" + self.assertEqual(len(domain), MAXIMUM_DOMAIN_LENGTH + 1) + + with self.assertRaises(OwidError) as raised: + Owid.from_byte_array(envelope(terminated(domain))) + + self.assertIn( + "'{0}'".format(MAXIMUM_DOMAIN_LENGTH), str(raised.exception) + ) + + def test_missing_terminator_is_refused_within_the_bound(self) -> None: + """A domain field with no terminator at all is refused, and the cost + of refusing it is set by the maximum rather than by the size of the + buffer. + + The buffer holds 64 MiB of domain characters and no zero byte + anywhere. A loop of 1,000 parses must finish in under a second, + which fails if each parse walks the whole buffer, and tracemalloc + must report a peak under 64 KiB for a single parse.""" + raw = hostile(b"a" * HOSTILE_LENGTH) + + started = time.perf_counter() + for _ in range(1000): + with self.assertRaises(OwidError): + Owid.from_byte_array(raw) + elapsed = time.perf_counter() - started + + self.assertLess( + elapsed, + 1.0, + "1,000 parses of a {0} byte buffer took {1:.3f}s".format( + HOSTILE_LENGTH, elapsed + ), + ) + tracemalloc.start() + try: + with self.assertRaises(OwidError): + Owid.from_byte_array(raw) + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + self.assertLess( + peak, + 64 * 1024, + "refusing the buffer reached a peak of {0} bytes".format(peak), + ) + + def test_distant_terminator_does_not_size_the_domain(self) -> None: + """A terminator far beyond the maximum is refused without building a + string of everything before it. + + Without the bound this is the most expensive case of the two, + because the reader would decode 64 MiB of domain characters into a + string rather than only walking past them. The peak must stay under + 64 KiB, which is a thousandth of what that string alone would + take.""" + raw = hostile(b"a" * HOSTILE_LENGTH + b"\0") + + tracemalloc.start() + try: + with self.assertRaises(OwidError): + Owid.from_byte_array(raw) + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + + self.assertLess( + peak, + 64 * 1024, + "refusing the buffer reached a peak of {0} bytes".format(peak), + ) + + def test_library_output_parses(self) -> None: + """A round trip through the library's own signing path and writer + still parses and still verifies, so the bound refuses nothing the + library itself produces.""" + crypto = Crypto.new() + original = Creator(MAXIMUM_DOMAIN, crypto).sign_bytes(PAYLOAD) + + parsed = Owid.from_byte_array(original.as_byte_array()) + + self.assertEqual(parsed.domain, MAXIMUM_DOMAIN) + self.assertEqual(parsed, original) + self.assertTrue(parsed.verify_with_crypto(crypto, [])) + + +class CountingCrypto: + """Stands in for a Crypto instance and counts the times it is asked to + sign. + + A Creator uses only can_sign and sign_byte_array, so this records the + signing work done on the creator side and lets a test prove a refusal + arrived before any of it. + """ + + def __init__(self, crypto: Crypto) -> None: + self._crypto = crypto + self.sign_calls = 0 + + def can_sign(self) -> bool: + return self._crypto.can_sign() + + def sign_byte_array(self, data: bytes) -> bytes: + self.sign_calls += 1 + return self._crypto.sign_byte_array(data) + + +class DomainLengthWriteTests(unittest.TestCase): + def test_maximum_length_domain_is_written_and_parses_back(self) -> None: + """A domain of exactly the maximum length is written and the value + parses back unchanged, so the bound refuses nothing at or under the + maximum.""" + owid = Owid( + version=Version.VERSION3, + domain=MAXIMUM_DOMAIN, + payload=PAYLOAD, + signature=SIGNATURE, + ) + + parsed = Owid.from_byte_array(owid.as_byte_array()) + + self.assertEqual(parsed.domain, MAXIMUM_DOMAIN) + self.assertEqual(parsed.payload, PAYLOAD) + self.assertEqual(parsed.signature, SIGNATURE) + + def test_empty_domain_is_still_written(self) -> None: + """An empty domain is written as the terminator on its own, exactly + as it was before the bound, because the refusal is at the top of the + range and nothing else.""" + buffer = bytearray() + + write_string(buffer, "") + + self.assertEqual(bytes(buffer), b"\0") + + def test_creator_refuses_a_domain_over_the_maximum(self) -> None: + """A creator is refused the domain when the caller supplies it, and + the message names the maximum so the caller can see which limit the + domain crossed.""" + domain = MAXIMUM_DOMAIN + "c" + self.assertEqual(len(domain), MAXIMUM_DOMAIN_LENGTH + 1) + + with self.assertRaises(OwidError) as raised: + Creator(domain, Crypto.new()) + + self.assertIn( + "'{0}'".format(MAXIMUM_DOMAIN_LENGTH), str(raised.exception) + ) + + def test_writing_a_domain_over_the_maximum_is_refused(self) -> None: + """A domain that arrived by a route other than the creator, here set + on the OWID directly, is refused when it is serialised, so the + library never emits an OWID it would refuse to read.""" + domain = MAXIMUM_DOMAIN + "c" + owid = Owid( + version=Version.VERSION3, + domain=domain, + payload=PAYLOAD, + signature=SIGNATURE, + ) + + with self.assertRaises(OwidError) as raised: + owid.as_byte_array() + + self.assertIn( + "'{0}'".format(MAXIMUM_DOMAIN_LENGTH), str(raised.exception) + ) + + def test_refusal_comes_before_any_signature_is_computed(self) -> None: + """The refusal arrives before any signing work, because signing a + value that will be refused is wasted work on the creator side. + + The counter proves it counts by signing once with a domain of the + maximum length, and then the creator with the longer domain is + refused without the counter moving again.""" + counting = CountingCrypto(Crypto.new()) + + Creator(MAXIMUM_DOMAIN, counting).sign_bytes(PAYLOAD) + self.assertEqual(counting.sign_calls, 1) + + with self.assertRaises(OwidError): + Creator(MAXIMUM_DOMAIN + "c", counting) + + self.assertEqual(counting.sign_calls, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_io.py b/tests/test_io.py index f86cf96..9b21e71 100644 --- a/tests/test_io.py +++ b/tests/test_io.py @@ -87,8 +87,12 @@ def test_u32_little_endian(self) -> None: self.assertEqual(reader.read_u32(), 0x0A242B01) def test_byte_array_round_trip(self) -> None: + # The signature follows the byte array because the only counted byte + # array in an OWID is the payload, and the reader checks that the + # declared length leaves exactly the signature after it. buffer = bytearray() io.write_byte_array(buffer, b"payload") + io.write_signature(buffer, bytes(io.SIGNATURE_LENGTH)) reader = io.Reader(bytes(buffer)) self.assertEqual(reader.read_byte_array(), b"payload") @@ -115,6 +119,12 @@ def test_read_string_without_terminator_raises(self) -> None: with self.assertRaises(OwidError): reader.read_string() + def test_string_scan_does_not_depend_on_the_payload_size(self) -> None: + payload = b"x" * (1024 * 1024) + reader = io.Reader(b"example.com\0" + payload) + self.assertEqual(reader.read_string(), "example.com") + self.assertEqual(reader.read_bytes(len(payload)), payload) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_payload_length.py b/tests/test_payload_length.py new file mode 100644 index 0000000..676e8c7 --- /dev/null +++ b/tests/test_payload_length.py @@ -0,0 +1,173 @@ +# **************************************************************************** +# Copyright 2026 51 Degrees Mobile Experts Limited (51degrees.com) +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# **************************************************************************** +"""Tests for the payload length check in the OWID parse. + +The payload length field of an OWID is whatever the sender declared, so +parsing must check it against the bytes present before sizing anything by +it. These tests prove that a declared length that does not leave exactly the +signature after the payload is refused, that refusing it costs no time or +memory sized by the declared number, and that a correctly sized envelope +still parses. The 64 byte signature is the fixed tail every valid OWID ends +with. +""" + +from __future__ import annotations + +import struct +import time +import tracemalloc +import unittest + +from owid import SIGNATURE_LENGTH, Creator, Crypto, Owid, OwidError, Version + +DOMAIN = "51d.es" +PAYLOAD = bytes([0x5A]) * 37 +SIGNATURE = bytes([0x99]) * SIGNATURE_LENGTH + + +def envelope(declared_length: int, payload: bytes, signature: bytes) -> bytes: + """A version 3 envelope, being the version byte, the domain with its + terminator, four minute bytes, the declared payload length, the payload + bytes given and the signature bytes given, so a test can make the + declared length and the bytes present disagree.""" + buffer = bytearray() + buffer.append(Version.VERSION3.as_byte()) + buffer.extend(DOMAIN.encode("ascii")) + buffer.append(0) + buffer.extend(struct.pack(" None: + """The declared length matches the bytes present, the signature is + the last 64 bytes, and the envelope parses to the same payload.""" + owid = Owid.from_byte_array( + envelope(len(PAYLOAD), PAYLOAD, SIGNATURE) + ) + self.assertEqual(owid.payload, PAYLOAD) + self.assertEqual(owid.signature, SIGNATURE) + self.assertEqual(owid.domain, DOMAIN) + + def test_matching_one_mebibyte_payload_parses(self) -> None: + """A matching large payload is valid; size policy belongs upstream.""" + payload = b"\x5a" * (1024 * 1024) + + owid = Owid.from_byte_array( + envelope(len(payload), payload, SIGNATURE) + ) + + self.assertEqual(owid.payload, payload) + + def test_library_output_parses(self) -> None: + """A round trip through the library's own signing path and writer + still parses, so the check agrees with what the library itself + produces.""" + crypto = Crypto.new() + original = Creator(DOMAIN, crypto).sign_bytes(PAYLOAD) + parsed = Owid.from_byte_array(original.as_byte_array()) + self.assertEqual(parsed.payload, PAYLOAD) + self.assertEqual(parsed, original) + self.assertTrue(parsed.verify_with_crypto(crypto, [])) + + def test_declared_length_off_by_one_is_refused(self) -> None: + """One more or one fewer than the bytes present is refused, because + either leaves something other than exactly the signature at the + end. The message names the declared length and the bytes present + so the reader of a log can see which the sender got wrong.""" + present = len(PAYLOAD) + SIGNATURE_LENGTH + for declared in (len(PAYLOAD) - 1, len(PAYLOAD) + 1): + with self.subTest(declared=declared): + with self.assertRaises(OwidError) as raised: + Owid.from_byte_array( + envelope(declared, PAYLOAD, SIGNATURE) + ) + message = str(raised.exception) + self.assertIn("'{0}'".format(declared), message) + self.assertIn("'{0}'".format(present), message) + + def test_trailing_byte_after_signature_is_refused(self) -> None: + """A byte after the signature is refused, because the signature + must be the end of the envelope.""" + longer = envelope(len(PAYLOAD), PAYLOAD, SIGNATURE) + b"\x00" + with self.assertRaises(OwidError): + Owid.from_byte_array(longer) + + def test_short_signature_is_refused(self) -> None: + """A short signature is refused. The declared payload length is + right for the payload, but the bytes after it are fewer than a + signature.""" + with self.assertRaises(OwidError): + Owid.from_byte_array( + envelope(len(PAYLOAD), PAYLOAD, SIGNATURE[:-1]) + ) + + def test_mismatched_large_declaration_is_refused_without_allocating( + self, + ) -> None: + """A large declaration whose payload bytes are absent is cheap. + + The envelope is a few dozen bytes while declaring 64 MiB, then 2 GiB, + then the largest value the field can hold. The numeric values remain + valid when the matching payload is present. Python cannot ask for + the bytes allocated by + the thread the way the .NET reference does, so two checks stand + in. A loop of 1,000 parses must finish in under a second, which + fails if each parse allocates the declared size, and tracemalloc + must report a peak under 64 KiB for a single parse.""" + for declared in (64 * 1024 * 1024, 0x7FFFFFFF, 0xFFFFFFFF): + with self.subTest(declared=declared): + raw = envelope(declared, b"", b"") + started = time.perf_counter() + for _ in range(1000): + with self.assertRaises(OwidError): + Owid.from_byte_array(raw) + elapsed = time.perf_counter() - started + self.assertLess( + elapsed, + 1.0, + "declared {0} took {1:.3f}s for 1,000 parses".format( + declared, elapsed + ), + ) + tracemalloc.start() + try: + with self.assertRaises(OwidError): + Owid.from_byte_array(raw) + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + self.assertLess( + peak, + 64 * 1024, + "declared {0} reached a peak of {1} bytes".format( + declared, peak + ), + ) + + def test_empty_payload_parses(self) -> None: + """A declared length of zero followed by the 64 byte signature + parses, so the check does not refuse the smallest valid payload.""" + owid = Owid.from_byte_array(envelope(0, b"", SIGNATURE)) + self.assertEqual(owid.payload, b"") + self.assertEqual(owid.signature, SIGNATURE) + + +if __name__ == "__main__": + unittest.main()