From d499dace402b08fb63815be805311713af3d129a Mon Sep 17 00:00:00 2001 From: "Benedicto Franco Jr." Date: Sat, 7 Nov 2026 18:53:20 -0800 Subject: [PATCH 1/5] pb: port protowire-go's encoding/pb, a PB codec for dataclasses Fields are declared with Annotated[T, pb.Field(number, kind)]. marshal, unmarshal and the length-prefixed Encoder / Decoder write and read protowire-go v1.17.0's bytes, big numbers included, under Go's per-call limits. Go's encoding/pb tests are ported with their byte vectors; the conformance tests check the bytes against the protobuf runtime over protoc-compiled schemas, both ways. --- src/protowire/__init__.py | 6 +- src/protowire/pb.py | 1246 +++++++++++++++++++++++++++ testdata/pb/bignum.proto | 25 + testdata/pb/conf.proto | 42 + tests/conftest.py | 11 + tests/test_pb.py | 666 ++++++++++++++ tests/test_pb_bigfloat.py | 211 +++++ tests/test_pb_bignum_conformance.py | 344 ++++++++ tests/test_pb_conformance.py | 297 +++++++ tests/test_pb_depth.py | 112 +++ tests/test_pb_limits.py | 683 +++++++++++++++ tests/test_pb_stream.py | 139 +++ 12 files changed, 3779 insertions(+), 3 deletions(-) create mode 100644 src/protowire/pb.py create mode 100644 testdata/pb/bignum.proto create mode 100644 testdata/pb/conf.proto create mode 100644 tests/test_pb.py create mode 100644 tests/test_pb_bigfloat.py create mode 100644 tests/test_pb_bignum_conformance.py create mode 100644 tests/test_pb_conformance.py create mode 100644 tests/test_pb_depth.py create mode 100644 tests/test_pb_limits.py create mode 100644 tests/test_pb_stream.py diff --git a/src/protowire/__init__.py b/src/protowire/__init__.py index ea98592..5ebac6d 100644 --- a/src/protowire/__init__.py +++ b/src/protowire/__init__.py @@ -1,8 +1,8 @@ # SPDX-License-Identifier: MIT # Copyright (c) 2026 TrendVidia, LLC. -"""protowire — PXF/SBE/envelope codecs, Python wrapper around protowire-cpp.""" +"""protowire — PXF/SBE/envelope codecs, Python wrapper around protowire-cpp, and a PB codec for dataclasses.""" -from . import envelope, pxf, sbe +from . import envelope, pb, pxf, sbe -__all__ = ["pxf", "sbe", "envelope"] +__all__ = ["pxf", "sbe", "envelope", "pb"] __version__ = "1.1.0" diff --git a/src/protowire/pb.py b/src/protowire/pb.py new file mode 100644 index 0000000..f78a57c --- /dev/null +++ b/src/protowire/pb.py @@ -0,0 +1,1246 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 TrendVidia, LLC. +"""Protobuf binary for plain Python classes — a port of protowire-go's encoding/pb. + +No ``.proto`` and no generated code: a dataclass declares each field's number +(and, where the Python type does not say it, the protobuf type) with +``typing.Annotated``:: + + from dataclasses import dataclass, field + from typing import Annotated, Optional + from protowire import pb + + @dataclass + class Endpoint: + path: Annotated[str, pb.Field(1)] = "" + port: Annotated[int, pb.Field(3, pb.INT32)] = 0 + delta: Annotated[int, pb.Field(4, pb.SINT64)] = 0 # Go's `protowire:"4,zigzag"` + tags: Annotated[list[str], pb.Field(5)] = field(default_factory=list) + inner: Annotated[Optional[Inner], pb.Field(6)] = None + + data = pb.marshal(Endpoint(path="/x", port=8080)) + back = pb.unmarshal(data, Endpoint) + +The bytes are protowire-go's, byte for byte, and follow proto3: + +- A zero singular field is omitted. ``None`` is an absent field, as Go's nil + pointer is; a value under ``Optional`` follows the same zero rule. +- A nested dataclass field that is not ``Optional`` is always written, as a Go + struct value is. +- Every element of a list and both halves of a map entry are written, zeros + included. A list of numbers (``int``, ``float``, ``bool``) is packed; + ``unmarshal`` reads packed and unpacked. +- A field is read by its declared type, whatever wire type the record carries, + as Go reads it. +- A ``str`` must be UTF-8 on the wire; declare ``bytes`` for a bytes field. +- ``unmarshal`` into an existing object merges, as Go's does: scalars are + replaced, lists are appended to, maps gain entries, and nested messages are + merged into. + +Types and their protobuf types: + +================================== ========================================== +Python protobuf +================================== ========================================== +``int`` int64; or INT32, UINT32, UINT64, SINT32, + SINT64, or BIGINT (``pxf.BigInt``) +``float`` double; or FLOAT +``bool``, ``str``, ``bytes`` bool, string, bytes +``decimal.Decimal`` ``pxf.Decimal`` +``pb.BigFloat`` ``pxf.BigFloat`` +a dataclass a message +``list[T]`` repeated T +``dict[K, V]`` map +``Optional[T]`` T, with ``None`` for absent +================================== ========================================== + +A protobuf type given in ``pb.Field`` applies to a list's elements or a map's +values. A map key that is not int64 says so on the key type: +``dict[Annotated[int, pb.UINT32], str]``. An ``int`` outside its declared +type's range is a ``ValueError`` from ``marshal``. + +Decoding is bounded by protowire's mandatory limits (HARDENING.md), each a +keyword argument of ``unmarshal`` and ``Decoder`` named after the Go option, +with 0 for the default. ``marshal`` refuses a value nested past +``MAX_NESTING_DEPTH``, since ``unmarshal`` would refuse it. + +Not ported: Go's ``UnmarshalOptions.Validator``, since this package has no +validation engine. A value nested past Python's recursion limit is a +``ValueError`` here; Go has no such ceiling, so a ``max_nesting_depth`` raised +past a few hundred needs ``sys.setrecursionlimit`` raised with it. +""" + +from __future__ import annotations + +import dataclasses +import decimal +import struct +import sys +import types +import typing +from dataclasses import dataclass +from fractions import Fraction +from typing import Annotated, Any, BinaryIO, Optional, TypeVar, Union + +__all__ = [ + "BIGINT", + "DEFAULT_MAX_FRAME_SIZE", + "FLOAT", + "INT32", + "INT64", + "MAX_BIG_FLOAT_EXPONENT", + "MAX_MESSAGE_SIZE", + "MAX_NESTING_DEPTH", + "MAX_NUMERIC_LITERAL_DIGITS", + "MAX_REPEATED_COUNT", + "SINT32", + "SINT64", + "UINT32", + "UINT64", + "DOUBLE", + "BigFloat", + "Decoder", + "Encoder", + "Field", + "FramingError", + "Kind", + "marshal", + "unmarshal", +] + +# --- Limits (protowire/docs/HARDENING.md § Mandatory limits) --------------- + +#: Submessage / map-entry recursion; the root is depth 0. +MAX_NESTING_DEPTH = 100 +#: Bounds |Decimal.scale| and the digits a BigFloat's prec implies. +MAX_NUMERIC_LITERAL_DIGITS = 4096 +#: Bounds |exponent + mantissa bit length| of a BigFloat. +MAX_BIG_FLOAT_EXPONENT = 16384 +#: Caps the input to one decode, checked before anything is read. +MAX_MESSAGE_SIZE = 64 << 20 +#: Caps the element count of any repeated or map field. +MAX_REPEATED_COUNT = MAX_MESSAGE_SIZE +#: The stream decoder's default frame cap. +DEFAULT_MAX_FRAME_SIZE = 16 << 20 + +_INT32_MIN, _INT32_MAX = -(1 << 31), (1 << 31) - 1 +_INT64_MIN, _INT64_MAX = -(1 << 63), (1 << 63) - 1 +_UINT32_MAX, _UINT64_MAX = (1 << 32) - 1, (1 << 64) - 1 +_MAX_FIELD_NUMBER = (1 << 29) - 1 +# protobuf-go's protowire.DefaultRecursionLimit, which bounds the groups an +# unknown field may nest. +_GROUP_RECURSION_LIMIT = 10000 + +_VARINT, _FIXED64, _BYTES, _START_GROUP, _END_GROUP, _FIXED32 = 0, 1, 2, 3, 4, 5 + + +# --- Declarations ---------------------------------------------------------- + + +@dataclass(frozen=True) +class Kind: + """A protobuf type that a Python type alone does not name.""" + + name: str + + def __repr__(self) -> str: + return f"pb.{self.name.upper()}" + + +INT32 = Kind("int32") +INT64 = Kind("int64") +UINT32 = Kind("uint32") +UINT64 = Kind("uint64") +SINT32 = Kind("sint32") +SINT64 = Kind("sint64") +BIGINT = Kind("bigint") +FLOAT = Kind("float") +DOUBLE = Kind("double") + +_INT_KINDS = {INT32, INT64, UINT32, UINT64, SINT32, SINT64, BIGINT} +_FLOAT_KINDS = {FLOAT, DOUBLE} + + +@dataclass(frozen=True) +class Field: + """The field number, and optionally the protobuf type, of a dataclass field. + + Used as ``Annotated[T, pb.Field(number)]`` or + ``Annotated[T, pb.Field(number, pb.SINT64)]``. + """ + + number: int + kind: Optional[Kind] = None + + +@dataclass(frozen=True, eq=True) +class BigFloat: + """A ``pxf.BigFloat``: ``(-1)**negative * mantissa * 2**exponent`` at ``prec`` bits. + + It holds a value as protowire-go's ``big.Float`` holds it, and does no + arithmetic. The fields are normalised on construction: + + - ``mantissa`` is rounded to ``prec`` bits (to nearest, ties to even), then + reduced to an odd number, with ``exponent`` adjusted to keep the value; + - a zero is ``mantissa=0, exponent=0`` and keeps its ``prec`` and sign; + - ``prec=0`` holds only a signed zero; + - a value whose binary exponent (``exponent + mantissa.bit_length()``) + leaves the int32 range is an error above it and a zero below it, as for + ``big.Float``. + + ``marshal`` writes the mantissa widened to ``prec`` bits, as Go does. + """ + + mantissa: int = 0 + exponent: int = 0 + prec: int = 0 + negative: bool = False + + def __post_init__(self) -> None: + m, e, p = self.mantissa, self.exponent, self.prec + if not isinstance(m, int) or not isinstance(e, int) or not isinstance(p, int): + raise TypeError("BigFloat mantissa, exponent and prec are ints") + if m < 0: + raise ValueError("BigFloat mantissa is a magnitude; the sign is `negative`") + if not 0 <= p <= _UINT32_MAX: + raise ValueError(f"BigFloat prec {p} is outside uint32") + if p == 0 or m == 0: + m, e = 0, 0 + else: + m, e = _round_to_prec(m, e, p) + binexp = e + m.bit_length() + if binexp > _INT32_MAX: + raise ValueError(f"BigFloat binary exponent {binexp} overflows big.Float") + if binexp < _INT32_MIN: + m, e = 0, 0 # big.Float underflows to a signed zero + else: + tz = (m & -m).bit_length() - 1 + m, e = m >> tz, e + tz + object.__setattr__(self, "mantissa", m) + object.__setattr__(self, "exponent", e) + object.__setattr__(self, "negative", bool(self.negative)) + + def to_fraction(self) -> Fraction: + """The exact value. A signed zero is ``Fraction(0)``.""" + v = Fraction(self.mantissa) * (Fraction(2) ** self.exponent) + return -v if self.negative else v + + +def _round_to_prec(m: int, e: int, prec: int) -> tuple[int, int]: + """m * 2**e rounded to prec bits, ties to even, as big.Float.SetInt rounds.""" + n = m.bit_length() + if n <= prec: + return m, e + shift = n - prec + q = m >> shift + r = m & ((1 << shift) - 1) + half = 1 << (shift - 1) + if r > half or (r == half and q & 1): + q += 1 + if q.bit_length() > prec: + q >>= 1 + shift += 1 + return q, e + shift + + +# --- Schema ---------------------------------------------------------------- + +# A resolved type is a tuple whose first element says what it is: +# ("scalar", kind) kind: a Kind, or "bool", "string", "bytes", +# "decimal", "bigfloat" +# ("message", cls) +# ("optional", T) +# ("list", T) +# ("map", K, V) +_T = tuple + + +@dataclass +class _FieldInfo: + name: str + number: int + type: _T + + +@dataclass +class _StructInfo: + cls: type + fields: list[_FieldInfo] + by_number: dict[int, _FieldInfo] + + +_struct_cache: dict[type, _StructInfo] = {} + +_NONE_TYPE = type(None) + + +def _is_union(tp: Any) -> bool: + return typing.get_origin(tp) is Union or isinstance(tp, types.UnionType) + + +def _split_optional(tp: Any) -> tuple[Any, bool]: + if _is_union(tp): + args = [a for a in typing.get_args(tp) if a is not _NONE_TYPE] + if len(args) == 1 and len(typing.get_args(tp)) == 2: + return args[0], True + raise TypeError(f"pb: unsupported union {tp!r}; only Optional[T] is") + return tp, False + + +def _annotated(tp: Any) -> tuple[Any, tuple]: + if typing.get_origin(tp) is Annotated: + return tp.__origin__, tp.__metadata__ + return tp, () + + +def _find_field(tp: Any) -> tuple[Any, Optional[Field]]: + """The type and the pb.Field of a dataclass field's annotation, if any. + + Python 3.10's get_type_hints wraps a None-defaulted annotation in + Optional, Annotated and all, so the Field is looked for there too. + """ + inner, meta = _annotated(tp) + fields = [m for m in meta if isinstance(m, Field)] + if fields: + return inner, fields[-1] + if _is_union(tp): + args = typing.get_args(tp) + if len(args) == 2 and _NONE_TYPE in args: + other = args[0] if args[1] is _NONE_TYPE else args[1] + inner, meta = _annotated(other) + fields = [m for m in meta if isinstance(m, Field)] + if fields: + return Optional[inner], fields[-1] + return tp, None + + +def _resolve(tp: Any, kind: Optional[Kind], where: str, *, top: bool = True) -> _T: + tp, meta = _annotated(tp) + for m in meta: + if isinstance(m, Kind): + kind = m + tp, optional = _split_optional(tp) + if optional: + return ("optional", _resolve(tp, kind, where, top=top)) + origin = typing.get_origin(tp) + if origin is list: + if not top: + raise TypeError(f"pb: {where}: a list inside a list or map is not supported") + (elem,) = typing.get_args(tp) or (None,) + if elem is None: + raise TypeError(f"pb: {where}: list needs an element type") + return ("list", _resolve(elem, kind, where, top=False)) + if origin is dict: + if not top: + raise TypeError(f"pb: {where}: a map inside a list or map is not supported") + args = typing.get_args(tp) + if len(args) != 2: + raise TypeError(f"pb: {where}: dict needs key and value types") + key = _resolve(args[0], None, where, top=False) + k = key[1] if key[0] == "scalar" else None + if key[0] != "scalar" or k in ("bytes", "decimal", "bigfloat", BIGINT, FLOAT, DOUBLE): + raise TypeError(f"pb: {where}: a map key must be an int, bool or str") + return ("map", key, _resolve(args[1], kind, where, top=False)) + if tp is bool: + _no_kind(kind, tp, where) + return ("scalar", "bool") + if tp is int: + if kind is None: + kind = INT64 + if kind not in _INT_KINDS: + raise TypeError(f"pb: {where}: {kind!r} is not an int type") + return ("scalar", kind) + if tp is float: + if kind is None: + kind = DOUBLE + if kind not in _FLOAT_KINDS: + raise TypeError(f"pb: {where}: {kind!r} is not a float type") + return ("scalar", kind) + if tp is str: + _no_kind(kind, tp, where) + return ("scalar", "string") + if tp is bytes: + _no_kind(kind, tp, where) + return ("scalar", "bytes") + if tp is decimal.Decimal: + _no_kind(kind, tp, where) + return ("scalar", "decimal") + if tp is BigFloat: + _no_kind(kind, tp, where) + return ("scalar", "bigfloat") + if isinstance(tp, type) and dataclasses.is_dataclass(tp): + _no_kind(kind, tp, where) + return ("message", tp) + raise TypeError(f"pb: {where}: unsupported type {tp!r}") + + +def _no_kind(kind: Optional[Kind], tp: Any, where: str) -> None: + if kind is not None: + raise TypeError(f"pb: {where}: {kind!r} does not apply to {tp.__name__}") + + +def _struct_info(cls: type) -> _StructInfo: + info = _struct_cache.get(cls) + if info is not None: + return info + if not (isinstance(cls, type) and dataclasses.is_dataclass(cls)): + raise TypeError(f"pb: expected a dataclass, got {cls!r}") + hints = typing.get_type_hints(cls, include_extras=True) + fields: list[_FieldInfo] = [] + by_number: dict[int, _FieldInfo] = {} + for f in dataclasses.fields(cls): + tp, pf = _find_field(hints.get(f.name, f.type)) + if pf is None: + continue + where = f"{cls.__name__}.{f.name}" + if not isinstance(pf.number, int) or not 1 <= pf.number <= _MAX_FIELD_NUMBER: + raise TypeError(f"pb: {where}: field number {pf.number!r} is outside 1..{_MAX_FIELD_NUMBER}") + if pf.number in by_number: + raise TypeError(f"pb: {where}: field number {pf.number} is also {by_number[pf.number].name}") + fi = _FieldInfo(f.name, pf.number, _resolve(tp, pf.kind, where)) + fields.append(fi) + by_number[pf.number] = fi + info = _StructInfo(cls, fields, by_number) + _struct_cache[cls] = info + return info + + +def _zero(t: _T) -> Any: + tag = t[0] + if tag == "optional": + return None + if tag == "list": + return [] + if tag == "map": + return {} + if tag == "message": + return _new(t[1]) + k = t[1] + if k == "bool": + return False + if k == "string": + return "" + if k == "bytes": + return b"" + if k == "decimal": + return decimal.Decimal(0) + if k == "bigfloat": + return BigFloat() + if k in _FLOAT_KINDS: + return 0.0 + return 0 + + +def _new(cls: type) -> Any: + """A zero instance: every pb field without a default gets its type's zero.""" + info = _struct_info(cls) + pbtypes = {fi.name: fi.type for fi in info.fields} + kwargs = {} + for f in dataclasses.fields(cls): + if not f.init: + continue + if f.default is not dataclasses.MISSING or f.default_factory is not dataclasses.MISSING: + continue + t = pbtypes.get(f.name) + kwargs[f.name] = _zero(t) if t is not None else None + return cls(**kwargs) + + +# --- Encoding -------------------------------------------------------------- + + +def _put_varint(out: bytearray, v: int) -> None: + while v >= 0x80: + out.append((v & 0x7F) | 0x80) + v >>= 7 + out.append(v) + + +def _put_tag(out: bytearray, num: int, typ: int) -> None: + _put_varint(out, (num << 3) | typ) + + +def _put_bytes(out: bytearray, num: int, b: Union[bytes, bytearray]) -> None: + _put_tag(out, num, _BYTES) + _put_varint(out, len(b)) + out += b + + +def _depth_err(limit: int) -> ValueError: + return ValueError(f"nesting depth exceeds MaxNestingDepth={limit}") + + +def marshal(obj: Any, *, max_nesting_depth: int = 0) -> bytes: + """Encode a dataclass instance as protobuf binary. + + A value nested deeper than ``max_nesting_depth`` (0: ``MAX_NESTING_DEPTH``) + is an error, since ``unmarshal`` would refuse it. + """ + limit = max_nesting_depth if max_nesting_depth > 0 else MAX_NESTING_DEPTH + if not dataclasses.is_dataclass(obj) or isinstance(obj, type): + raise TypeError(f"pb.marshal: expected a dataclass instance, got {type(obj).__name__}") + try: + return bytes(_marshal_struct(obj, 0, limit)) + except RecursionError: + raise ValueError(_recursion_msg()) from None + + +def _recursion_msg() -> str: + return ( + f"message nests deeper than Python's recursion limit ({sys.getrecursionlimit()}) " + "allows; raise sys.setrecursionlimit" + ) + + +def _marshal_struct(obj: Any, depth: int, limit: int) -> bytearray: + if depth > limit: + raise _depth_err(limit) + info = _struct_info(type(obj)) + out = bytearray() + for fi in info.fields: + try: + _marshal_field(out, fi.number, fi.type, getattr(obj, fi.name), False, depth, limit) + except ValueError as e: + raise ValueError(f"field {fi.name}: {e}") from None + except TypeError as e: + raise TypeError(f"field {fi.name}: {e}") from None + return out + + +def _is_packed(t: _T) -> bool: + return t[0] == "scalar" and (t[1] == "bool" or t[1] in _FLOAT_KINDS or (t[1] in _INT_KINDS and t[1] is not BIGINT)) + + +def _marshal_field(out: bytearray, num: int, t: _T, v: Any, element: bool, depth: int, limit: int) -> None: + # Optional is Go's pointer: None is absent as a singular field, and an + # element written as its type's zero so the list keeps its length. + if t[0] == "optional": + t = t[1] + if v is None: + if not element: + return + v = _zero(t) + elif v is None: + if element: + raise TypeError("None element in a list or map not declared Optional") + return + tag = t[0] + if tag == "list": + if not isinstance(v, (list, tuple)): + raise TypeError(f"expected a list, got {type(v).__name__}") + if not v: + return + et = t[1] + if _is_packed(et): + payload = bytearray() + for e in v: + _put_scalar_value(payload, et[1], e) + _put_bytes(out, num, payload) + return + for e in v: + _marshal_field(out, num, et, e, True, depth, limit) + return + if tag == "map": + if not isinstance(v, dict): + raise TypeError(f"expected a dict, got {type(v).__name__}") + if not v: + return + if depth + 1 > limit: + raise _depth_err(limit) + for k, val in v.items(): + entry = bytearray() + try: + _marshal_field(entry, 1, t[1], k, True, depth + 1, limit) + except ValueError as e: + raise ValueError(f"map key: {e}") from None + try: + _marshal_field(entry, 2, t[2], val, True, depth + 1, limit) + except ValueError as e: + raise ValueError(f"map value: {e}") from None + _put_bytes(out, num, entry) + return + if tag == "message": + if not isinstance(v, t[1]): + raise TypeError(f"expected {t[1].__name__}, got {type(v).__name__}") + _put_bytes(out, num, _marshal_struct(v, depth + 1, limit)) + return + _marshal_scalar(out, num, t[1], v, element) + + +def _check_int(k: Kind, v: Any) -> int: + if not isinstance(v, int): + raise TypeError(f"expected an int, got {type(v).__name__}") + v = int(v) + if k is INT32 or k is SINT32: + lo, hi = _INT32_MIN, _INT32_MAX + elif k is INT64 or k is SINT64: + lo, hi = _INT64_MIN, _INT64_MAX + elif k is UINT32: + lo, hi = 0, _UINT32_MAX + elif k is UINT64: + lo, hi = 0, _UINT64_MAX + else: + return v + if not lo <= v <= hi: + raise ValueError(f"{v} is outside {k.name}") + return v + + +def _float_bits(k: Kind, v: Any) -> bytes: + if not isinstance(v, (int, float)) or isinstance(v, bool): + raise TypeError(f"expected a float, got {type(v).__name__}") + v = float(v) + # A NaN is written as the canonical quiet NaN whatever its payload, as Go + # writes it (draft -01 § Numeric Literals; protowire-java#161). + if k is DOUBLE: + return b"\x00\x00\x00\x00\x00\x00\xf8\x7f" if v != v else struct.pack(" int: + if k is SINT32 or k is SINT64: + return ((v << 1) ^ (v >> 63)) & _UINT64_MAX # protobuf-go's EncodeZigZag + return v & _UINT64_MAX # int32/int64 negatives: sign-extended to 64 bits + + +def _put_scalar_value(out: bytearray, k: Any, v: Any) -> None: + """One packed element: its value encoding, no tag.""" + if k == "bool": + if not isinstance(v, bool): + raise TypeError(f"expected a bool, got {type(v).__name__}") + out.append(1 if v else 0) + elif k in _FLOAT_KINDS: + out += _float_bits(k, v) + else: + _put_varint(out, _varint_of(k, _check_int(k, v))) + + +def _marshal_scalar(out: bytearray, num: int, k: Any, v: Any, element: bool) -> None: + if k == "bool": + if not isinstance(v, bool): + raise TypeError(f"expected a bool, got {type(v).__name__}") + if not v and not element: + return + _put_tag(out, num, _VARINT) + out.append(1 if v else 0) + elif k == "string": + if not isinstance(v, str): + raise TypeError(f"expected a str, got {type(v).__name__}") + if not v and not element: + return + try: + b = v.encode("utf-8") + except UnicodeEncodeError as e: + raise ValueError(f"string is not valid UTF-8: {e}") from None + _put_bytes(out, num, b) + elif k == "bytes": + if not isinstance(v, (bytes, bytearray, memoryview)): + raise TypeError(f"expected bytes, got {type(v).__name__}") + b = bytes(v) + if not b and not element: + return + _put_bytes(out, num, b) + elif k == "decimal": + if not isinstance(v, decimal.Decimal): + raise TypeError(f"expected a decimal.Decimal, got {type(v).__name__}") + if not v.is_finite(): + raise ValueError(f"cannot encode {v} as pxf.Decimal") + if v.is_zero() and not element: + return + _put_bytes(out, num, _marshal_decimal(v)) + elif k == "bigfloat": + if not isinstance(v, BigFloat): + raise TypeError(f"expected a pb.BigFloat, got {type(v).__name__}") + # Only +0 at prec 0 is the empty message; -0 there is + # {negative: true}, as PXF "-0" reads (protowire#410). + if v.prec == 0 and not v.negative and not element: + return + _put_bytes(out, num, _marshal_bigfloat(v)) + elif k is BIGINT: + v = _check_int(k, v) + if v == 0 and not element: + return + _put_bytes(out, num, _marshal_bigint(v)) + elif k in _FLOAT_KINDS: + bits = _float_bits(k, v) + if v == 0 and not element: + return + _put_tag(out, num, _FIXED64 if k is DOUBLE else _FIXED32) + out += bits + else: + v = _check_int(k, v) + if v == 0 and not element: + return + _put_tag(out, num, _VARINT) + _put_varint(out, _varint_of(k, v)) + + +def _int_bytes(n: int) -> bytes: + return n.to_bytes((n.bit_length() + 7) // 8, "big") + + +def _marshal_bigint(v: int) -> bytearray: + msg = bytearray() + if v != 0: + _put_bytes(msg, 1, _int_bytes(abs(v))) + if v < 0: + _put_tag(msg, 2, _VARINT) + msg.append(1) + return msg + + +# Exact decimal arithmetic: nothing here may round. +_DEC_CTX = decimal.Context( + prec=decimal.MAX_PREC, + Emax=decimal.MAX_EMAX, + Emin=decimal.MIN_EMIN, + traps=[decimal.InvalidOperation, decimal.Overflow, decimal.Inexact, decimal.Rounded], +) + + +def _marshal_decimal(v: decimal.Decimal) -> bytearray: + # Go holds a Decimal as a big.Rat, which keeps no trailing zeros, and + # writes the smallest scale that makes the value an integer, never a + # negative one. normalize() is that form. + if v.is_zero(): + unscaled, scale = 0, 0 + else: + n = v.copy_abs().normalize(_DEC_CTX) + exp = n.as_tuple().exponent + if exp >= 0: + unscaled, scale = int(n), 0 + else: + scale = -exp + unscaled = int(n.scaleb(scale, _DEC_CTX)) + # Go's own Unmarshal refuses a scale past the bound, so writing one would + # produce bytes no conformant decoder reads (protowire-go#95). + if scale > MAX_NUMERIC_LITERAL_DIGITS: + raise ValueError( + f"pxf.Decimal scale {scale} exceeds MaxNumericLiteralDigits={MAX_NUMERIC_LITERAL_DIGITS}" + ) + msg = bytearray() + if unscaled: + _put_bytes(msg, 1, _int_bytes(unscaled)) + if scale: + _put_tag(msg, 2, _VARINT) + _put_varint(msg, scale) + if unscaled and v.is_signed(): + _put_tag(msg, 3, _VARINT) + msg.append(1) + return msg + + +def _pow2_digits(p: int) -> int: + """The decimal digits of 2**p, floor(p*log10(2)) + 1, without computing 2**p. + + The 128-bit fixed-point product protowire-go's internal/bigprec uses, exact + for every uint32 p (float64 is one digit high at p = 146964308). + """ + return ((p * 0x4D104D427DE7FBCC47C4ACD605BE48BC) >> 128) + 1 + + +def _prec_digits(prec: int) -> int: + """N, the significant digits a BigFloat of prec bits is written with.""" + return _pow2_digits(prec) + 1 + + +def _exponent_err(mant_bits: int, exponent: int, limit: int) -> Optional[ValueError]: + if mant_bits == 0: + return None + e = exponent + mant_bits + if e > limit or e < -limit: + return ValueError(f"pxf.BigFloat binary exponent {e} exceeds MaxBigFloatExponent={limit}") + return None + + +def _marshal_bigfloat(v: BigFloat) -> bytearray: + msg = bytearray() + prec = v.prec + if prec == 0: + if v.negative: + _put_tag(msg, 4, _VARINT) + msg.append(1) + return msg + n = _prec_digits(prec) + if n > MAX_NUMERIC_LITERAL_DIGITS: + raise ValueError( + f"pxf.BigFloat prec {prec} implies {n} digits; MaxNumericLiteralDigits={MAX_NUMERIC_LITERAL_DIGITS}" + ) + # The wire mantissa is the value scaled to exactly prec bits. + m, adj = v.mantissa, 0 + if m: + widen = prec - m.bit_length() + m <<= widen + adj = v.exponent - widen + if not _INT32_MIN <= adj <= _INT32_MAX: + raise ValueError(f"pxf.BigFloat exponent {adj} does not fit int32") + err = _exponent_err(m.bit_length(), adj, MAX_BIG_FLOAT_EXPONENT) + if err is not None: + raise err + if m: + _put_bytes(msg, 1, _int_bytes(m)) + if adj: + _put_tag(msg, 2, _VARINT) + _put_varint(msg, adj & _UINT64_MAX) + _put_tag(msg, 3, _VARINT) + _put_varint(msg, prec) + if v.negative: + _put_tag(msg, 4, _VARINT) + msg.append(1) + return msg + + +# --- Decoding -------------------------------------------------------------- + + +@dataclass(frozen=True) +class _Limits: + max_depth: int + max_digits: int + max_big_float_exp: int + max_message_size: int + max_repeated: int + + +def _limits( + max_message_size: int, + max_nesting_depth: int, + max_numeric_literal_digits: int, + max_repeated_count: int, + max_big_float_exponent: int, +) -> _Limits: + def pick(v: int, default: int) -> int: + return v if v > 0 else default + + return _Limits( + max_depth=pick(max_nesting_depth, MAX_NESTING_DEPTH), + max_digits=pick(max_numeric_literal_digits, MAX_NUMERIC_LITERAL_DIGITS), + max_big_float_exp=pick(max_big_float_exponent, MAX_BIG_FLOAT_EXPONENT), + max_message_size=pick(max_message_size, MAX_MESSAGE_SIZE), + max_repeated=pick(max_repeated_count, MAX_REPEATED_COUNT), + ) + + +class _Corrupt(Exception): + """A wire-level failure; the caller names what was being read.""" + + +def _varint(buf: bytes, pos: int, end: int) -> tuple[int, int]: + v = 0 + shift = 0 + for i in range(10): + if pos >= end: + raise _Corrupt + b = buf[pos] + pos += 1 + if i == 9: + if b >= 2: # protobuf-go: the 10th byte carries one bit + raise _Corrupt + return v | (b << 63), pos + v |= (b & 0x7F) << shift + if b < 0x80: + return v, pos + shift += 7 + raise AssertionError("unreachable") + + +def _tag(buf: bytes, pos: int, end: int) -> tuple[int, int, int]: + v, pos = _varint(buf, pos, end) + num = v >> 3 + if num > _INT32_MAX or num < 1: + raise _Corrupt + return num, v & 7, pos + + +def _len_prefixed(buf: bytes, pos: int, end: int) -> tuple[int, int]: + """The start and end of a length-prefixed value.""" + n, pos = _varint(buf, pos, end) + if n > end - pos: + raise _Corrupt + return pos, pos + n + + +def _fixed(buf: bytes, pos: int, end: int, size: int) -> int: + if end - pos < size: + raise _Corrupt + return pos + size + + +def _skip(num: int, typ: int, buf: bytes, pos: int, end: int) -> int: + """protobuf-go's ConsumeFieldValue: the position after one field's value.""" + if typ == _VARINT: + return _varint(buf, pos, end)[1] + if typ == _FIXED64: + return _fixed(buf, pos, end, 8) + if typ == _FIXED32: + return _fixed(buf, pos, end, 4) + if typ == _BYTES: + return _len_prefixed(buf, pos, end)[1] + if typ != _START_GROUP: + raise _Corrupt # a stray end-group, or a reserved wire type + groups = [num] + while groups: + n2, t2, pos = _tag(buf, pos, end) + if t2 == _END_GROUP: + if n2 != groups[-1]: + raise _Corrupt + groups.pop() + elif t2 == _START_GROUP: + if len(groups) > _GROUP_RECURSION_LIMIT: + raise _Corrupt + groups.append(n2) + else: + pos = _skip(n2, t2, buf, pos, end) + return pos + + +C = TypeVar("C") + + +def unmarshal( + data: Union[bytes, bytearray, memoryview], + obj: Union[C, type[C]], + *, + max_message_size: int = 0, + max_nesting_depth: int = 0, + max_numeric_literal_digits: int = 0, + max_repeated_count: int = 0, + max_big_float_exponent: int = 0, +) -> C: + """Decode protobuf binary into a dataclass instance, and return it. + + ``obj`` is an instance, which is merged into as Go's ``Unmarshal`` merges, + or a dataclass, of which a zero instance is made. Each limit is + protowire-go's ``UnmarshalOptions`` field of the same name; 0 selects the + default. A failure is a ``ValueError``. + """ + lim = _limits( + max_message_size, max_nesting_depth, max_numeric_literal_digits, max_repeated_count, max_big_float_exponent + ) + return _unmarshal(data, obj, lim) + + +def _unmarshal(data: Any, obj: Any, lim: _Limits) -> Any: + buf = bytes(data) + if len(buf) > lim.max_message_size: + raise ValueError(f"input of {len(buf)} bytes exceeds MaxMessageSize={lim.max_message_size}") + if isinstance(obj, type): + obj = _new(obj) + elif not dataclasses.is_dataclass(obj): + raise TypeError(f"pb.unmarshal: expected a dataclass or an instance of one, got {type(obj).__name__}") + try: + _unmarshal_struct(buf, 0, len(buf), obj, 0, lim) + except RecursionError: + raise ValueError(_recursion_msg()) from None + return obj + + +def _unmarshal_struct(buf: bytes, pos: int, end: int, obj: Any, depth: int, lim: _Limits) -> None: + if depth > lim.max_depth: + raise _depth_err(lim.max_depth) + info = _struct_info(type(obj)) + by_number = info.by_number + while pos < end: + try: + num, typ, pos = _tag(buf, pos, end) + except _Corrupt: + raise ValueError("corrupt tag") from None + fi = by_number.get(num) + if fi is None: + try: + pos = _skip(num, typ, buf, pos, end) + except _Corrupt: + raise ValueError(f"corrupt field {num}") from None + continue + try: + value, pos = _unmarshal_field(buf, pos, end, typ, fi.type, getattr(obj, fi.name), depth, lim) + except ValueError as e: + raise ValueError(f"field {fi.name}: {e}") from None + setattr(obj, fi.name, value) + + +def _unmarshal_field( + buf: bytes, pos: int, end: int, typ: int, t: _T, cur: Any, depth: int, lim: _Limits +) -> tuple[Any, int]: + """Read one record's value by its declared type; return (new value, position).""" + if t[0] == "optional": + t = t[1] + tag = t[0] + if tag == "list": + lst = cur if cur is not None else [] + et = t[1] + bare = et[1] if et[0] == "optional" else et + if typ == _BYTES and _is_packed(bare): + try: + p, pend = _len_prefixed(buf, pos, end) + except _Corrupt: + raise ValueError("corrupt packed field") from None + while p < pend: + v, p = _unmarshal_field(buf, p, pend, typ, et, None, depth, lim) + if len(lst) >= lim.max_repeated: + raise ValueError(f"repeated field exceeds MaxRepeatedCount={lim.max_repeated}") + lst.append(v) + return lst, pend + v, pos = _unmarshal_field(buf, pos, end, typ, et, None, depth, lim) + if len(lst) >= lim.max_repeated: + raise ValueError(f"repeated field exceeds MaxRepeatedCount={lim.max_repeated}") + lst.append(v) + return lst, pos + if tag == "message": + try: + p, pend = _len_prefixed(buf, pos, end) + except _Corrupt: + raise ValueError("corrupt embedded message") from None + msg = cur if cur is not None else _new(t[1]) + _unmarshal_struct(buf, p, pend, msg, depth + 1, lim) + return msg, pend + if tag == "map": + try: + p, pend = _len_prefixed(buf, pos, end) + except _Corrupt: + raise ValueError("corrupt map entry") from None + # The entry is a descent whatever its value's type (HARDENING.md + # § Recursion; protowire#449). + if depth + 1 > lim.max_depth: + raise _depth_err(lim.max_depth) + d = cur if cur is not None else {} + kt, vt = t[1], t[2] + key, val = _zero(kt), _zero(vt) + while p < pend: + try: + enum, etyp, p = _tag(buf, p, pend) + except _Corrupt: + raise ValueError("corrupt tag in map entry") from None + if enum == 1: + try: + key, p = _unmarshal_field(buf, p, pend, etyp, kt, key, depth + 1, lim) + except ValueError as e: + raise ValueError(f"map key: {e}") from None + elif enum == 2: + try: + val, p = _unmarshal_field(buf, p, pend, etyp, vt, val, depth + 1, lim) + except ValueError as e: + raise ValueError(f"map value: {e}") from None + else: + try: + p = _skip(enum, etyp, buf, p, pend) + except _Corrupt: + raise ValueError(f"corrupt unknown field {enum} in map entry") from None + if len(d) >= lim.max_repeated and key not in d: + raise ValueError(f"map field exceeds MaxRepeatedCount={lim.max_repeated}") + d[key] = val + return d, pend + return _unmarshal_scalar(buf, pos, end, t[1], lim) + + +def _signed(v: int, bits: int) -> int: + v &= (1 << bits) - 1 + return v - (1 << bits) if v >> (bits - 1) else v + + +def _unmarshal_scalar(buf: bytes, pos: int, end: int, k: Any, lim: _Limits) -> tuple[Any, int]: + if k in _FLOAT_KINDS: + size = 8 if k is DOUBLE else 4 + try: + p = _fixed(buf, pos, end, size) + except _Corrupt: + raise ValueError(f"corrupt {'float64' if k is DOUBLE else 'float32'}") from None + return struct.unpack_from("> 1) ^ -(v & 1) # zigzag, as protobuf-go's DecodeZigZag + return (_signed(z, 64) if k is SINT64 else _signed(z, 32)), p + + +def _sub_fields(buf: bytes, pos: int, end: int, what: str, varints: set, blobs: set) -> dict: + """The last value of each known field of a big-number message.""" + out: dict = {} + while pos < end: + try: + num, typ, pos = _tag(buf, pos, end) + except _Corrupt: + raise ValueError(f"corrupt tag in {what}") from None + try: + if num in blobs: + p, pos = _len_prefixed(buf, pos, end) + out[num] = buf[p:pos] + elif num in varints: + out[num], pos = _varint(buf, pos, end) + else: + pos = _skip(num, typ, buf, pos, end) + except _Corrupt: + raise ValueError(f"corrupt field {num} in {what}") from None + return out + + +def _unmarshal_bigint(buf: bytes, pos: int, end: int) -> int: + f = _sub_fields(buf, pos, end, "BigInt", {2}, {1}) + v = int.from_bytes(f.get(1, b""), "big") + return -v if f.get(2, 0) else v + + +def _unmarshal_decimal(buf: bytes, pos: int, end: int, max_digits: int) -> decimal.Decimal: + f = _sub_fields(buf, pos, end, "Decimal", {2, 3}, {1}) + unscaled = int.from_bytes(f.get(1, b""), "big") + scale = _signed(f.get(2, 0), 32) # int32: the low 32 bits, read as signed + # value = unscaled x 10^(-scale), and both signs of scale materialise + # 10^|scale|, so the bound comes first (protowire-go#95). + if scale > max_digits or scale < -max_digits: + raise ValueError(f"pxf.Decimal scale {scale} exceeds MaxNumericLiteralDigits={max_digits}") + d = decimal.Decimal(unscaled).scaleb(-scale, _DEC_CTX) + # A big.Rat has no negative zero. + return -d if f.get(3, 0) and unscaled else d + + +def _unmarshal_bigfloat(buf: bytes, pos: int, end: int, max_exp: int, max_digits: int) -> BigFloat: + f = _sub_fields(buf, pos, end, "BigFloat", {2, 3, 4}, {1}) + negative = bool(f.get(4, 0)) + prec = f.get(3, 0) & _UINT32_MAX # uint32: the low 32 bits (protowire-go#230) + if prec == 0: + # A zero at prec 0, signed by negative (protowire#410). + return BigFloat(0, 0, 0, negative) + n = _prec_digits(prec) + if n > max_digits: + raise ValueError(f"pxf.BigFloat prec {prec} implies {n} digits; MaxNumericLiteralDigits={max_digits}") + exp = _signed(f.get(2, 0), 32) + mant = int.from_bytes(f.get(1, b""), "big") + err = _exponent_err(mant.bit_length(), exp, max_exp) + if err is not None: + raise err + if mant: + q, e = _round_to_prec(mant, exp, prec) + if e + q.bit_length() > _INT32_MAX: + raise ValueError( + f"pxf.BigFloat exponent {exp} with a {mant.bit_length()}-bit mantissa overflows big.Float" + ) + return BigFloat(mant, exp, prec, negative) + + +# --- Streams --------------------------------------------------------------- + + +class FramingError(ValueError): + """The length-prefix framing of a stream is invalid. + + The stream is out of step after this, and the Decoder stays failed. + """ + + +class Encoder: + """Writes length-prefixed messages: a varint length, then the body.""" + + def __init__(self, w: BinaryIO, *, max_nesting_depth: int = 0) -> None: + self._w = w + self._max_depth = max_nesting_depth + + def encode(self, obj: Any) -> None: + body = marshal(obj, max_nesting_depth=self._max_depth) + hdr = bytearray() + _put_varint(hdr, len(body)) + self._w.write(bytes(hdr)) + self._w.write(body) + + +class Decoder: + """Reads length-prefixed messages written by ``Encoder`` or protowire-go's. + + ``decode`` returns ``None`` at a clean end of stream. A truncated frame + raises ``EOFError`` and corrupt framing ``FramingError``; both leave the + Decoder failed. A frame whose message does not decode raises + ``ValueError`` and the next frame can still be read. + """ + + def __init__( + self, + r: BinaryIO, + *, + max_frame_size: int = DEFAULT_MAX_FRAME_SIZE, + max_message_size: int = 0, + max_nesting_depth: int = 0, + max_numeric_literal_digits: int = 0, + max_repeated_count: int = 0, + max_big_float_exponent: int = 0, + ) -> None: + self._r = r + self._max = max_frame_size + self._err: Optional[Exception] = None + self._lim = _limits( + max_message_size, max_nesting_depth, max_numeric_literal_digits, max_repeated_count, max_big_float_exponent + ) + + def _fail(self, e: Exception) -> Exception: + self._err = e + return e + + def decode(self, obj: Union[C, type[C]]) -> Optional[C]: + if self._err is not None: + raise self._err + # encoding/binary.ReadUvarint: ten bytes at most, the tenth one bit. + n = 0 + shift = 0 + for i in range(10): + b = self._r.read(1) + if not b: + if i == 0: + return None + raise self._fail(EOFError("unexpected EOF")) + c = b[0] + if c < 0x80: + if i == 9 and c > 1: + raise self._fail(FramingError("pb: stream framing corrupt: varint overflows a 64-bit integer")) + n |= c << shift + break + n |= (c & 0x7F) << shift + shift += 7 + else: + raise self._fail(FramingError("pb: stream framing corrupt: varint overflows a 64-bit integer")) + if n > self._max: + raise self._fail(FramingError(f"pb: stream framing corrupt: frame size {n} exceeds max {self._max}")) + body = bytearray() + while len(body) < n: + chunk = self._r.read(n - len(body)) + if not chunk: + raise self._fail(EOFError("unexpected EOF")) + body += chunk + return _unmarshal(body, obj, self._lim) diff --git a/testdata/pb/bignum.proto b/testdata/pb/bignum.proto new file mode 100644 index 0000000..3d9d6cc --- /dev/null +++ b/testdata/pb/bignum.proto @@ -0,0 +1,25 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 TrendVidia, LLC. +// +// pxf/bignum.proto, as protowire-go's encoding/pb/bignum_conformance_test.go +// states it: the oracle schema for the big-number messages. +syntax = "proto3"; +package pxf; + +message BigInt { + bytes abs = 1; + bool negative = 2; +} + +message Decimal { + bytes unscaled = 1; + int32 scale = 2; + bool negative = 3; +} + +message BigFloat { + bytes mantissa = 1; + int32 exponent = 2; + uint32 prec = 3; + bool negative = 4; +} diff --git a/testdata/pb/conf.proto b/testdata/pb/conf.proto new file mode 100644 index 0000000..5e3c240 --- /dev/null +++ b/testdata/pb/conf.proto @@ -0,0 +1,42 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 TrendVidia, LLC. +// +// The oracle schema for protowire.pb's conformance tests: protowire-go's +// encoding/pb/conformance_test.go conformanceProtoSrc, verbatim. +syntax = "proto3"; +package conf; + +message Inner { + string name = 1; + int64 n = 2; +} + +message All { + string s = 1; + int64 i64 = 2; + int32 i32 = 3; + uint64 u64 = 4; + uint32 u32 = 5; + sint64 z64 = 6; + sint32 z32 = 7; + bool b = 8; + double f64 = 9; + float f32 = 10; + bytes buf = 11; + Inner inner = 12; + repeated Inner list = 13; + repeated uint32 nums = 14; + repeated sint64 zigs = 15; + repeated float reals = 16; + repeated string strs = 17; + map m = 18; + map mm = 19; + oneof choice { + string a = 20; + int64 c = 21; + Inner d = 22; + } + repeated bool flags = 23; + repeated int64 negs = 24; + map bm = 25; +} diff --git a/tests/conftest.py b/tests/conftest.py index 7f64822..2008512 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -82,3 +82,14 @@ def all_types_cls(test_pool): def nested_cls(test_pool): desc = test_pool.FindMessageTypeByName("test.v1.Nested") return message_factory.GetMessageClass(desc) + + +@pytest.fixture(scope="session") +def pb_oracle_pool() -> descriptor_pool.DescriptorPool: + """The oracle schemas for protowire.pb: testdata/pb/conf.proto and bignum.proto.""" + fds = descriptor_pb2.FileDescriptorSet() + fds.ParseFromString(_protoc_to_fds(["conf.proto", "bignum.proto"], [str(TESTDATA_DIR / "pb")])) + pool = descriptor_pool.DescriptorPool() + for f in fds.file: + pool.Add(f) + return pool diff --git a/tests/test_pb.py b/tests/test_pb.py new file mode 100644 index 0000000..d81b045 --- /dev/null +++ b/tests/test_pb.py @@ -0,0 +1,666 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 TrendVidia, LLC. +"""protowire.pb against protowire-go's encoding/pb tests (v1.17.0, ca79f5c). + +Ports pb_test.go, utf8_test.go and example_test.go. Each test names the Go +test it ports. Where a Go test only round-trips, the port also checks the +bytes against what protowire-go's ``Marshal`` writes for the same value, +measured by running v1.17.0 (those vectors are marked "Go Marshal bytes"). + +Error messages name the Python attribute where Go names the Go struct field +(``field name: ...`` for Go's ``field Name: ...``); the rest of each message is +Go's text. +""" + +from __future__ import annotations + +import io +import math +import struct +from dataclasses import dataclass, field +from decimal import Decimal +from typing import Annotated, Optional + +import pytest + +from protowire import pb + +# --- wire helpers: google.golang.org/protobuf/encoding/protowire's Append* --- + +_VARINT, _FIXED32, _BYTES = 0, 5, 2 + + +def append_varint(b: bytes, v: int) -> bytes: + out = bytearray(b) + v &= (1 << 64) - 1 + while v >= 0x80: + out.append((v & 0x7F) | 0x80) + v >>= 7 + out.append(v) + return bytes(out) + + +def append_tag(b: bytes, num: int, typ: int) -> bytes: + return append_varint(b, (num << 3) | typ) + + +def append_bytes(b: bytes, v: bytes) -> bytes: + return append_varint(b, len(v)) + v + + +def len_field(b: bytes, num: int, v: bytes) -> bytes: + return append_bytes(append_tag(b, num, _BYTES), v) + + +def big_float_of(x: float, prec: int) -> pb.BigFloat: + """new(big.Float).SetPrec(prec).SetFloat64(x).""" + n, d = abs(x).as_integer_ratio() + return pb.BigFloat(n, -(d.bit_length() - 1), prec, x < 0) + + +# --- pb_test.go -------------------------------------------------------------- + + +@dataclass +class Inner: + name: Annotated[str, pb.Field(1)] = "" + value: Annotated[int, pb.Field(2)] = 0 + + +@dataclass +class Outer: + title: Annotated[str, pb.Field(1)] = "" + count: Annotated[int, pb.Field(2, pb.UINT32)] = 0 + score: Annotated[float, pb.Field(3)] = 0.0 + active: Annotated[bool, pb.Field(4)] = False + data: Annotated[bytes, pb.Field(5)] = b"" + items: Annotated[list[Inner], pb.Field(6)] = field(default_factory=list) + ptrs: Annotated[list[Optional[Inner]], pb.Field(7)] = field(default_factory=list) + signed: Annotated[int, pb.Field(8)] = 0 + small_f: Annotated[float, pb.Field(9, pb.FLOAT)] = 0.0 + untagged: str = "" # no pb.Field — should be skipped + + +def test_round_trip(): + """Ports TestRoundTrip.""" + orig = Outer( + title="hello", + count=42, + score=3.14, + active=True, + data=b"\xde\xad", + items=[Inner("a", 1), Inner("b", -7)], + ptrs=[Inner("p", 99)], + signed=-12345, + small_f=2.5, + untagged="should be ignored", + ) + data = pb.marshal(orig) + # Go Marshal bytes. + assert data.hex() == ( + "0a0568656c6c6f102a191f85eb51b81e094020012a02dead32050a01611001320e0a0162" + "10f9ffffffffffffffff013a050a0170106340c79fffffffffffffff014d00002040" + ) + got = pb.unmarshal(data, Outer()) + # Untagged field should not round-trip + orig.untagged = "" + assert got == orig + + +def test_zero_values(): + """Ports TestZeroValues.""" + data = pb.marshal(Outer()) + # Proto3: all-zero struct should produce empty bytes + assert data == b"" + pb.unmarshal(data, Outer()) + + +# Go's TestCustomByteType declares `type CustomByte byte`: a field of Go kind +# uint8. pb has no uint8 type (Python has no named integer kinds), and Go +# writes every unsigned kind as the same plain varint, so the nearest port +# declares the field UINT32. What it can no longer test is the Go-only part: +# that a named type over byte is not mistaken for []byte's element. +@dataclass +class WithCustomByte: + tag: Annotated[int, pb.Field(1, pb.UINT32)] = 0 + value: Annotated[int, pb.Field(2)] = 0 + + +def test_custom_byte_type(): + """Ports TestCustomByteType (as UINT32; see the comment above).""" + orig = WithCustomByte(tag=3, value=-42) + data = pb.marshal(orig) + assert data.hex() == "080310d6ffffffffffffffff01" # Go Marshal bytes + assert pb.unmarshal(data, WithCustomByte) == orig + + +@dataclass +class Big3: + a: Annotated[str, pb.Field(1)] = "" + b: Annotated[str, pb.Field(2)] = "" + c: Annotated[str, pb.Field(3)] = "" + + +@dataclass +class Small1: + a: Annotated[str, pb.Field(1)] = "" + + +def test_unknown_fields_skipped(): + """Ports TestUnknownFieldsSkipped.""" + data = pb.marshal(Big3(a="aa", b="bb", c="cc")) + # Decode into a smaller struct — field 2 and 3 are unknown + got = pb.unmarshal(data, Small1) + assert got.a == "aa" + + +@dataclass +class BigNumStruct: + balance: Annotated[Optional[int], pb.Field(1, pb.BIGINT)] = None + price: Annotated[Optional[Decimal], pb.Field(2)] = None + coefficient: Annotated[Optional[pb.BigFloat], pb.Field(3)] = None + + +def test_big_num_round_trip(): + """Ports TestBigNumRoundTrip.""" + balance = int("115792089237316195423570985008687907853269984665640564039457584007913129639935") + price = Decimal("3.1415") # new(big.Rat).SetFrac64(31415, 10000) + coeff = big_float_of(6.02214076e23, 128) + orig = BigNumStruct(balance=balance, price=price, coefficient=coeff) + + data = pb.marshal(orig) + assert data.hex() == ( # Go Marshal bytes + "0a220a20ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" + "12060a027ab710041a200a10ff0c2e52be28b800000000000000000010cfffffffffffffffff01188001" + ) + got = pb.unmarshal(data, BigNumStruct(balance=0, price=Decimal(0), coefficient=pb.BigFloat())) + assert got.balance == orig.balance + assert got.price == orig.price + assert got.coefficient.to_fraction() == orig.coefficient.to_fraction() + + +def test_big_num_nil_pointers(): + """Ports TestBigNumNilPointers.""" + assert pb.marshal(BigNumStruct()) == b"" + + +def test_big_num_negative(): + """Ports TestBigNumNegative. + + Go's Price is the big.Rat -1/3, which no decimal.Decimal holds. Go writes + a non-terminating rational as its 50-digit FloatString (ratToDecimal), so + the port gives the Decimal those 50 digits, and checks that the bytes are + the ones Go writes for -1/3. + """ + orig = BigNumStruct( + balance=-999999999999, + price=Decimal("-0." + "3" * 50), + coefficient=big_float_of(-1.23e-45, 64), + ) + data = pb.marshal(orig) + assert data.hex() == ( # Go Marshal bytes, Price = big.Rat -1/3 + "0a090a05e8d4a50fff1001121b0a1516cebe5ca886222d46415711a9e721555555555555103218011a190a08" + "e0b4b45687b3c00010abfeffffffffffffff0118402001" + ) + got = pb.unmarshal(data, BigNumStruct(balance=0, price=Decimal(0), coefficient=pb.BigFloat())) + assert got.balance == orig.balance + # For non-terminating decimal, we lose precision — just check sign + assert got.price < 0 + assert got.coefficient.to_fraction() == orig.coefficient.to_fraction() + + +def test_big_num_pointer_fields(): + """Ports TestBigNumPointerFields.""" + data = pb.marshal(BigNumStruct(balance=42)) + assert data.hex() == "0a030a012a" # Go Marshal bytes + got = pb.unmarshal(data, BigNumStruct(balance=0)) + assert got.balance == 42 + # A None field is allocated by unmarshal, as Go allocates a nil pointer. + assert pb.unmarshal(data, BigNumStruct).balance == 42 + + +@dataclass +class WithZigZag: + a: Annotated[int, pb.Field(1)] = 0 # proto3 int64 (plain varint) + b: Annotated[int, pb.Field(2, pb.SINT64)] = 0 # proto3 sint64 (zigzag varint) + + +def test_zig_zag_tag_option(): + """Ports TestZigZagTagOption.""" + orig = WithZigZag(a=-1, b=-1) + data = pb.marshal(orig) + # Field 1 (A=-1, plain varint): tag 0x08, then 10 bytes of 0xff..0x01. + # Field 2 (B=-1, zigzag): tag 0x10, then varint(zigzag(-1)) = varint(1) = 0x01. + assert len(data) == 13, data.hex() + assert data.hex() == "08ffffffffffffffffff011001" # Go Marshal bytes + assert pb.unmarshal(data, WithZigZag) == orig + + +@dataclass +class WithMaps: + tags: Annotated[dict[str, str], pb.Field(1)] = field(default_factory=dict) + codes: Annotated[dict[Annotated[int, pb.INT32], str], pb.Field(2)] = field(default_factory=dict) + counts: Annotated[dict[str, int], pb.Field(3)] = field(default_factory=dict) + servers: Annotated[dict[str, Inner], pb.Field(4)] = field(default_factory=dict) + + +def test_map_round_trip(): + """Ports TestMapRoundTrip.""" + orig = WithMaps( + tags={"env": "prod", "team": "platform", "key with space": "v"}, + codes={404: "Not Found", 500: "Internal", -1: "Negative"}, + counts={"a": 1, "b": -7, "c": 0}, + servers={"primary": Inner("p", 1), "secondary": Inner("s", 2)}, + ) + data = pb.marshal(orig) + assert pb.unmarshal(data, WithMaps) == orig + + +def test_map_entry_bytes(): + """Go Marshal bytes for one entry per map (Go's map order is random, so + TestMapRoundTrip's values cannot be compared byte for byte).""" + v = WithMaps(tags={"env": "prod"}, codes={-1: "Negative"}, counts={"c": 0}, servers={"primary": Inner("p", 1)}) + assert pb.marshal(v).hex() == ( + "0a0b0a03656e76120470726f64121508ffffffffffffffffff0112084e65676174697665" + "1a050a0163100022100a077072696d61727912050a01701001" + ) + + +@pytest.mark.parametrize( + "val", + [WithMaps(), WithMaps(tags={}, codes={})], + ids=["nil maps", "empty maps"], +) +def test_empty_map(val): + """Ports TestEmptyMap: empty and absent maps both produce empty bytes.""" + assert pb.marshal(val) == b"" + + +def test_map_zero_value_entries(): + """Ports TestMapZeroValueEntries. + + Go's comment says the zero key and value "are skipped per proto3 + semantics", but Go writes both halves of every entry, zeros included: + its bytes for this value are an entry ``0a 00 12 00`` (pb.go's map case + marshals key and value as elements). The port checks those bytes. + """ + orig = WithMaps(tags={"": "", "k": ""}, counts={"": 0}) + data = pb.marshal(orig) + got = pb.unmarshal(data, WithMaps) + assert got.tags == {"": "", "k": ""} + assert got.counts == {"": 0} + + one = WithMaps(tags={"": ""}, counts={"": 0}) + assert pb.marshal(one).hex() == "0a040a0012001a040a001000" # Go Marshal bytes + + +@dataclass +class WithPacked: + u32: Annotated[list[int], pb.Field(1, pb.UINT32)] = field(default_factory=list) + i64: Annotated[list[int], pb.Field(2)] = field(default_factory=list) + zig: Annotated[list[int], pb.Field(3, pb.SINT32)] = field(default_factory=list) + f32: Annotated[list[float], pb.Field(4, pb.FLOAT)] = field(default_factory=list) + f64: Annotated[list[float], pb.Field(5)] = field(default_factory=list) + bs: Annotated[list[bool], pb.Field(6)] = field(default_factory=list) + + +def test_packed_round_trip(): + """Ports TestPackedRoundTrip.""" + # Zero-valued elements must survive: packed encoding has no + # per-element presence, so nothing may be dropped. + orig = WithPacked( + u32=[1, 0, 2, 300], + i64=[-1, 0, 42], + zig=[-5, 0, 5], + f32=[1.5, 0, -2.25], + f64=[3.14, 0, -1e100], + bs=[True, False, True], + ) + data = pb.marshal(orig) + assert data.hex() == ( # Go Marshal bytes + "0a05010002ac02120cffffffffffffffffff01002a1a0309000a220c0000c03f00000000000010c0" + "2a181f85eb51b81e094000000000000000007dc39425ad49b2d43203010001" + ) + assert pb.unmarshal(data, WithPacked) == orig + + +def test_packed_wire_format(): + """Ports TestPackedWireFormat.""" + # Canonical example from the protobuf encoding docs: repeated + # int32/uint32 {3, 270, 86942} packs to `LEN 6: 03 8E 02 9E A7 05`. + data = pb.marshal(WithPacked(u32=[3, 270, 86942])) + assert data == bytes([0x0A, 0x06, 0x03, 0x8E, 0x02, 0x9E, 0xA7, 0x05]) + + +def test_unpacked_decode(): + """Ports TestUnpackedDecode.""" + # proto3 decoders must accept unpacked encodings of packable fields + # (one tag per element) — e.g. from proto2 or a non-default encoder. + data = b"" + for v in (7, 0, 9): + data = append_varint(append_tag(data, 1, _VARINT), v) + data = append_tag(data, 4, _FIXED32) + struct.pack(" bytes: + return len_field(len_field(b"", 1, key), 2, value) + + +_UTF8_SITES = [ + ("singular", lambda s: len_field(b"", 1, s), "field name: invalid UTF-8 in string"), + ("pointer", lambda s: len_field(b"", 2, s), "field ptr: invalid UTF-8 in string"), + ("named type", lambda s: len_field(b"", 3, s), "field named: invalid UTF-8 in string"), + ("list element", lambda s: len_field(len_field(b"", 4, b"ok"), 4, s), "field list: invalid UTF-8 in string"), + ("nested", lambda s: len_field(b"", 5, len_field(b"", 1, s)), "field inner: field text: invalid UTF-8 in string"), + ("map key", lambda s: len_field(b"", 6, _entry(s, b"v")), "field labels: map key: invalid UTF-8 in string"), + ("map value", lambda s: len_field(b"", 6, _entry(b"k", s)), "field labels: map value: invalid UTF-8 in string"), +] + +_UTF8_SEQUENCES = [ + ("stray byte", b"AB\xff"), + ("lone continuation", b"\x80"), + ("latin-1", b"caf\xe9"), + ("cut off", b"abc\xc3"), + ("overlong", b"\xc0\x80"), + ("surrogate", b"\xed\xa0\x80"), + ("past U+10FFFF", b"\xf4\x90\x80\x80"), +] + + +@pytest.mark.parametrize("site", _UTF8_SITES, ids=[s[0] for s in _UTF8_SITES]) +@pytest.mark.parametrize("seq", _UTF8_SEQUENCES, ids=[s[0] for s in _UTF8_SEQUENCES]) +def test_string_field_rejects_invalid_utf8(site, seq): + """Ports TestStringFieldRejectsInvalidUTF8. + + HARDENING.md § UTF-8 names PB length-delimited bytes. Each place a string + can sit, and each way a byte sequence can be invalid. + """ + _, wire_of, want = site + wire = wire_of(seq[1]) + with pytest.raises(ValueError) as e: + pb.unmarshal(wire, Utf8Msg()) + assert str(e.value) == want + + framed = append_bytes(b"", wire) + with pytest.raises(ValueError) as e: + pb.Decoder(io.BytesIO(framed)).decode(Utf8Msg()) + assert str(e.value) == want + assert not isinstance(e.value, pb.FramingError) + + +def test_string_field_accepts_valid_utf8(): + """Ports TestStringFieldAcceptsValidUTF8: the check must not refuse what the rule allows.""" + want = Utf8Msg( + name="é", + ptr="€", + named="😀", + list=["a\x00b", ""], + inner=Utf8Inner(text="日本語"), + labels={"ключ": "значение"}, + data=bytes([0xFF, 0xFE, 0x80, 0xC0]), + ) + wire = pb.marshal(want) + assert pb.unmarshal(wire, Utf8Msg) == want + + +def test_decoder_continues_past_invalid_utf8(): + """Ports TestDecoderContinuesPastInvalidUTF8: bad text is a bad message, not a bad stream.""" + stream = b"" + for text in (b"AB\xff", b"good"): + stream = append_bytes(stream, len_field(b"", 1, text)) + dec = pb.Decoder(io.BytesIO(stream)) + + with pytest.raises(ValueError) as e: + dec.decode(Utf8Msg()) + assert str(e.value) == "field name: invalid UTF-8 in string" + nxt = dec.decode(Utf8Msg()) + assert nxt is not None and nxt.name == "good" + assert dec.decode(Utf8Msg()) is None # Go: io.EOF + + +# --- example_test.go --------------------------------------------------------- + + +@dataclass +class Endpoint: + path: Annotated[str, pb.Field(1)] = "" + method: Annotated[str, pb.Field(2)] = "" + port: Annotated[int, pb.Field(3)] = 0 + + +@dataclass +class Config: + hostname: Annotated[str, pb.Field(1)] = "" + enabled: Annotated[bool, pb.Field(2)] = False + endpoints: Annotated[list[Optional[Endpoint]], pb.Field(3)] = field(default_factory=list) + + +def test_example(): + """Ports Example: encode and decode a struct with field-number tags. + + The output is standard protobuf binary and is wire-compatible with any + .proto message that uses the same field numbers and matching types. + """ + src = Config(hostname="web-01", enabled=True, endpoints=[Endpoint(path="/api", method="GET", port=8080)]) + data = pb.marshal(src) + assert data.hex() == "0a067765622d303110011a0e0a042f617069120347455418903f" # Go Marshal bytes + dst = pb.unmarshal(data, Config) + out = f"{dst.hostname} {str(dst.enabled).lower()} {dst.endpoints[0].path} {dst.endpoints[0].port}" + assert out == "web-01 true /api 8080" # Go's // Output: + + +@dataclass +class Tick: + symbol: Annotated[str, pb.Field(1)] = "" + price: Annotated[int, pb.Field(2)] = 0 + + +def test_example_encoder(): + """Ports ExampleEncoder: the length-prefixed stream API, messages back to back.""" + buf = io.BytesIO() + enc = pb.Encoder(buf) + enc.encode(Tick(symbol="AAPL", price=18900)) + enc.encode(Tick(symbol="GOOG", price=14250)) + # Go Marshal bytes for each Tick, each behind its varint length. + assert buf.getvalue().hex() == "0a" + "0a044141504c10d49301" + "09" + "0a04474f4f4710aa6f" + + buf.seek(0) + dec = pb.Decoder(buf) + lines = [] + while (t := dec.decode(Tick)) is not None: + lines.append(f"{t.symbol} {t.price}") + assert lines == ["AAPL 18900", "GOOG 14250"] # Go's // Output: diff --git a/tests/test_pb_bigfloat.py b/tests/test_pb_bigfloat.py new file mode 100644 index 0000000..8914e31 --- /dev/null +++ b/tests/test_pb_bigfloat.py @@ -0,0 +1,211 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 TrendVidia, LLC. +"""pb.BigFloat: zeros on the wire, and normalisation as big.Float holds a value. + +The zero tests port protowire-go's encoding/pb/bigfloat_zero_test.go. The +normalisation tests are this port's own: pb.BigFloat stands in for Go's +big.Float, so it must round (to nearest, ties to even), carry, underflow and +overflow where big.Float does. The rounding is checked against an exact model +in rationals rather than against pb's own arithmetic. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from fractions import Fraction +from typing import Annotated, Optional + +import pytest + +from protowire import pb + +INT32_MIN, INT32_MAX = -(1 << 31), (1 << 31) - 1 +BIG_MAX_PREC = (1 << 32) - 1 # Go's big.MaxPrec + + +@dataclass +class BigFloatHolder: + f: Annotated[Optional[pb.BigFloat], pb.Field(1)] = None + + +# --- bigfloat_zero_test.go ------------------------------------------------- + + +@pytest.mark.parametrize( + "wire,neg", [(bytes([0x0A, 0x00]), False), (bytes([0x0A, 0x02, 0x20, 0x01]), True)], ids=["zero", "negative zero"] +) +def test_prec_zero_keeps_its_sign_read(wire, neg): + """Ports TestBigFloatPrecZeroKeepsItsSign, the decode cases (protowire#410). + + A prec-0 big.Float is ±0, and PXF writes the two apart, "0" and "-0". The + bytes are the spec corpus's (testdata/bigfloat dec-zero and + dec-negative-zero): -0 is {negative: true} with no precision. Go used to + read it as +0 at 53 bits, so a "-0" that came through PB lost its sign. + """ + h = pb.unmarshal(wire, BigFloatHolder) + assert h.f.mantissa == 0 and h.f.prec == 0 and h.f.negative == neg, f"{wire.hex(' ')} reads as {h.f}" + + +def test_prec_zero_keeps_its_sign_write(): + """Ports TestBigFloatPrecZeroKeepsItsSign, the encode half: Go's Marshal used + to write nothing for a prec-0 -0.""" + out = pb.marshal(BigFloatHolder(f=pb.BigFloat(0, 0, 0, True))) # new(big.Float).Neg(new(big.Float)) + assert out == bytes([0x0A, 0x02, 0x20, 0x01]) + back = pb.unmarshal(out, BigFloatHolder) + assert back.f.negative and back.f.prec == 0, f"round trip: {back.f}; want -0 at prec 0" + # +0 at prec 0 stays the omitted field it was. + assert pb.marshal(BigFloatHolder(f=pb.BigFloat())) == b"" + + +@pytest.mark.parametrize( + "neg,want", + [(False, bytes([0x0A, 0x02, 0x18, 0x40])), (True, bytes([0x0A, 0x04, 0x18, 0x40, 0x20, 0x01]))], + ids=["zero", "negative zero"], +) +def test_zero_has_one_encoding(neg, want): + """Ports TestBigFloatZeroHasOneEncoding (protowire-go#196). + + A zero at any prec is written with an empty mantissa and exponent 0, + keeping prec and negative, which is what a PXF decoder stores for 0p64 + (draft -01 § PXF Annotations). The bytes are the spec corpus's + dec-zero-p64 and dec-negative-zero-p64. Go's Marshal used to write + exponent -prec, a second encoding of the same value. + """ + out = pb.marshal(BigFloatHolder(f=pb.BigFloat(0, 0, 64, neg))) + assert out == want, f"zero at prec 64 marshals to {out.hex(' ')}, want {want.hex(' ')}" + back = pb.unmarshal(out, BigFloatHolder) + assert back.f.mantissa == 0 and back.f.prec == 64 and back.f.negative == neg + + +def test_zero_mantissa_with_any_exponent_reads_as_zero(): + """Ports TestBigFloatZeroHasOneEncoding's last case: a zero mantissa with any + exponent still reads as zero, so bytes the old encoder wrote stay + readable.""" + old = bytes([0x0A, 0x0F, 0x10, 0xC0, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x01, 0x18, 0x40, 0x20, 0x01]) + h = pb.unmarshal(old, BigFloatHolder) + assert h.f.mantissa == 0 and h.f.prec == 64 and h.f.negative, f"{old.hex(' ')} reads as {h.f}; want -0 at prec 64" + + +@pytest.mark.parametrize("neg", [False, True]) +def test_zero_at_max_prec(neg): + """Ports TestBigFloatZeroAtMaxPrec (protowire-go#196, protowire#499). + + A zero writes no exponent, so its precision does not bound one: at prec + 2^32-1, exponent -prec left int32 and Go's Marshal refused the zero. Its + prec is still bounded, as any BigFloat's is: the zero at 2^32-1 bits is + refused on writing and on reading, and the zero at the bound, 13603 bits, + round-trips. + """ + with pytest.raises(ValueError, match="MaxNumericLiteralDigits"): + pb.marshal(BigFloatHolder(f=pb.BigFloat(0, 0, BIG_MAX_PREC, neg))) + # The bytes Go's Marshal used to write: prec 2^32-1, nothing else. + old = bytes([0x0A, 0x06, 0x18, 0xFF, 0xFF, 0xFF, 0xFF, 0x0F]) + if neg: + old = bytes([0x0A, 0x08, 0x18, 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, 0x20, 0x01]) + with pytest.raises(ValueError, match="MaxNumericLiteralDigits"): + pb.unmarshal(old, BigFloatHolder) + + out = pb.marshal(BigFloatHolder(f=pb.BigFloat(0, 0, 13603, neg))) + back = pb.unmarshal(out, BigFloatHolder) + assert back.f.mantissa == 0 and back.f.prec == 13603 and back.f.negative == neg, f"{out.hex(' ')} reads as {back.f}" + + +# --- Normalisation (this port's own) --------------------------------------- + + +def _model(m: int, e: int, prec: int) -> Fraction: + """m * 2**e rounded to prec bits, to nearest, ties to even, in rationals.""" + v = Fraction(m) * Fraction(2) ** e + if m == 0: + return v + k = m.bit_length() - prec # scale so the integer part has prec bits + if k <= 0: + return v + unit = Fraction(2) ** (e + k) + q, r = divmod(v, unit) + q = int(q) + if r > unit / 2 or (r == unit / 2 and q % 2 == 1): + q += 1 + return q * unit + + +def test_rounding_matches_the_model_exhaustively(): + """Every mantissa below 2^10 at every prec 1..11: rounded as big.Float + rounds, and stored reduced to an odd mantissa.""" + for prec in range(1, 12): + for m in range(1, 1 << 10): + for e in (-7, 0, 5): + bf = pb.BigFloat(m, e, prec) + assert bf.to_fraction() == _model(m, e, prec), (m, e, prec, bf) + assert bf.mantissa & 1 == 1, bf + assert bf.mantissa.bit_length() <= prec, bf + assert bf.prec == prec + + +@pytest.mark.parametrize( + "m,prec,mantissa,exponent", + [ + (0b101, 2, 1, 2), # 5 is between 4 and 6: the tie goes to even 4 + (0b1011, 3, 3, 2), # 11 is between 10 and 12: the tie goes to even 12 + (0b1001, 3, 1, 3), # 9 is between 8 and 10: the tie goes to even 8 + (0b1101, 2, 3, 2), # 13 is below the midpoint 14: down to 12 + (0b1111, 3, 1, 4), # 15 rounds up to 16, carrying into a new bit + (0b111, 2, 1, 3), # 7 ties between 6 and 8, goes to even 8: a carry + ], +) +def test_rounding_ties_to_even_and_carry(m, prec, mantissa, exponent): + bf = pb.BigFloat(m, 0, prec) + assert (bf.mantissa, bf.exponent) == (mantissa, exponent) + + +def test_negative_rounds_by_magnitude(): + """The sign is apart from the magnitude, so -5 at 2 bits is -4, as + big.Float's ToNearestEven rounds the magnitude.""" + bf = pb.BigFloat(5, 0, 2, True) + assert bf.to_fraction() == -4 + + +def test_prec_zero_holds_only_a_zero(): + bf = pb.BigFloat(5, 3, 0, True) + assert (bf.mantissa, bf.exponent, bf.prec, bf.negative) == (0, 0, 0, True) + + +def test_zero_keeps_prec_and_sign(): + bf = pb.BigFloat(0, 17, 64, True) + assert (bf.mantissa, bf.exponent, bf.prec, bf.negative) == (0, 0, 64, True) + + +def test_underflow_below_int32_is_a_zero(): + """big.Float's binary exponent is an int32; below MinInt32 a value + underflows to a zero of its sign, keeping prec.""" + at = pb.BigFloat(1, INT32_MIN - 1, 53) # binary exponent MinInt32 + assert (at.mantissa, at.exponent) == (1, INT32_MIN - 1) + below = pb.BigFloat(1, INT32_MIN - 2, 53, True) # binary exponent MinInt32 - 1 + assert (below.mantissa, below.exponent, below.prec, below.negative) == (0, 0, 53, True) + + +def test_overflow_above_int32_is_an_error(): + """Above MaxInt32 big.Float overflows to ±Inf. pb.BigFloat has no infinity, + so it refuses the value instead (a divergence of representation, not of + the wire: an infinity has no pxf.BigFloat encoding either).""" + at = pb.BigFloat(1, INT32_MAX - 1, 53) # binary exponent MaxInt32 + assert at.exponent == INT32_MAX - 1 + with pytest.raises(ValueError, match="overflows"): + pb.BigFloat(1, INT32_MAX, 53) + + +def test_rounding_carry_can_overflow(): + """3 x 2^(MaxInt32-2) has binary exponent MaxInt32; at 1 bit it rounds to + 2^MaxInt32, whose binary exponent is one past the range.""" + pb.BigFloat(3, INT32_MAX - 2, 2) + with pytest.raises(ValueError, match="overflows"): + pb.BigFloat(3, INT32_MAX - 2, 1) + + +def test_rejects_bad_fields(): + with pytest.raises(ValueError): + pb.BigFloat(-1, 0, 53) + with pytest.raises(ValueError): + pb.BigFloat(1, 0, 1 << 32) + with pytest.raises(TypeError): + pb.BigFloat(1.5, 0, 53) # type: ignore[arg-type] diff --git a/tests/test_pb_bignum_conformance.py b/tests/test_pb_bignum_conformance.py new file mode 100644 index 0000000..c537d68 --- /dev/null +++ b/tests/test_pb_bignum_conformance.py @@ -0,0 +1,344 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 TrendVidia, LLC. +"""Conformance for pb's big-number codec against an INDEPENDENT oracle. + +Ports protowire-go's encoding/pb/bignum_conformance_test.go (protowire-go#92). + +``pxf/bignum.proto`` declares ``Decimal.scale`` and ``BigFloat.exponent`` as +plain ``int32``, which is a plain varint on the wire; zigzag is ``sint32``, +which neither field is. protowire-go once used zigzag for both, in all four +places (two writers, two readers), so it round-tripped its own bytes perfectly +and disagreed with every schema-conformant reader. Self-consistency is what +hid it: a test that marshals with ``marshal`` and unmarshals with +``unmarshal`` can never catch it. + +The oracle here is google.protobuf over a descriptor compiled from +testdata/pb/bignum.proto itself (the ``pb_oracle_pool`` fixture), where Go +uses protobuf-go's dynamicpb. Both directions are covered: bytes this package +WRITES must be readable by a conformant decoder, and bytes a conformant +encoder WRITES must be readable here. + +Only zero survives the difference: over the full int32 range, zigzag and plain +varint agree on exactly one value, so any non-zero scale or exponent was wrong +on the wire, not merely the negative ones. +""" + +from __future__ import annotations + +import decimal +from dataclasses import dataclass +from fractions import Fraction +from typing import Annotated, Optional + +import pytest +from google.protobuf import message_factory + +from protowire import pb + +# --- Wire helpers: written here, not borrowed from pb, so the test does not +# read the bytes with the code under test. --------------------------------- + + +def _append_varint(out: bytearray, v: int) -> None: + v &= (1 << 64) - 1 + while v >= 0x80: + out.append((v & 0x7F) | 0x80) + v >>= 7 + out.append(v) + + +def _append_tag(out: bytearray, num: int, typ: int) -> None: + _append_varint(out, (num << 3) | typ) + + +def _append_bytes(out: bytearray, b: bytes) -> None: + _append_varint(out, len(b)) + out += b + + +def _consume_varint(b: bytes, pos: int) -> tuple[int, int]: + v, shift = 0, 0 + while True: + assert pos < len(b), "truncated varint" + c = b[pos] + pos += 1 + v |= (c & 0x7F) << shift + if c < 0x80: + return v, pos + shift += 7 + assert shift < 70, "varint overflows" + + +def _consume_field_value(typ: int, b: bytes, pos: int) -> int: + if typ == 0: + return _consume_varint(b, pos)[1] + if typ == 1: + return pos + 8 + if typ == 5: + return pos + 4 + if typ == 2: + n, pos = _consume_varint(b, pos) + assert pos + n <= len(b), "corrupt length" + return pos + n + raise AssertionError(f"unexpected wire type {typ}") + + +def _submessage(b: bytes, num: int) -> bytes: + """The length-delimited payload of field num in a record pb produced.""" + pos = 0 + while pos < len(b): + tag, pos = _consume_varint(b, pos) + n, typ = tag >> 3, tag & 7 + if n == num: + assert typ == 2 + ln, pos = _consume_varint(b, pos) + return b[pos : pos + ln] + pos = _consume_field_value(typ, b, pos) + pytest.fail(f"field {num} not present") + + +def _field_records(b: bytes) -> dict[int, bytes]: + """One entry per field, keyed by number, holding its bytes, tag included. + + Comparing these instead of the whole buffer keeps the assertion byte-exact + per field while not asserting an order the wire format does not fix (Go's + note: protobuf-go's output order is deliberately unstable, so a + whole-buffer comparison against it flaked ~13% of runs). + """ + out: dict[int, bytes] = {} + pos = 0 + while pos < len(b): + start = pos + tag, pos = _consume_varint(b, pos) + num, typ = tag >> 3, tag & 7 + pos = _consume_field_value(typ, b, pos) + assert num not in out, f"field {num} repeated" + out[num] = b[start:pos] + return out + + +def _wrap(num: int, sub: bytes) -> bytes: + """The outer record pb expects: one length-delimited field carrying sub.""" + out = bytearray() + _append_tag(out, num, 2) + _append_bytes(out, sub) + return bytes(out) + + +# --- An exact model of big.Float's rounding, for the expected values. ----- + + +def _round_fraction(v: Fraction, prec: int) -> pb.BigFloat: + """v rounded to prec bits, ties to even, as big.ParseFloat(.., ToNearestEven) + and big.Float.SetInt/SetMantExp round. Computed in exact rationals here, + independently of pb's own rounding.""" + neg = v < 0 + a = -v if neg else v + if a == 0 or prec == 0: + return pb.BigFloat(0, 0, prec, neg) + # e such that 2**(prec-1) <= a / 2**e < 2**prec. + e = a.numerator.bit_length() - a.denominator.bit_length() - prec + while a / Fraction(2) ** e >= 2**prec: + e += 1 + while a / Fraction(2) ** e < 2 ** (prec - 1): + e -= 1 + scaled = a / Fraction(2) ** e + q = scaled.numerator // scaled.denominator + r = scaled - q + if r > Fraction(1, 2) or (r == Fraction(1, 2) and q & 1): + q += 1 + return pb.BigFloat(q, e, prec, neg) + + +def _parse_float(text: str, prec: int) -> pb.BigFloat: + """big.ParseFloat(text, 10, prec, big.ToNearestEven).""" + return _round_fraction(Fraction(decimal.Decimal(text)), prec) + + +@dataclass +class ConformanceStruct: + """Go's conformanceStruct: the shape encoding/pb marshals.""" + + price: Annotated[Optional[decimal.Decimal], pb.Field(1)] = None + coefficient: Annotated[Optional[pb.BigFloat], pb.Field(2)] = None + + +@pytest.fixture(scope="module") +def oracle(pb_oracle_pool): + """The schema's Decimal and BigFloat classes (Go's bignumDescriptors).""" + get = lambda n: message_factory.GetMessageClass(pb_oracle_pool.FindMessageTypeByName(n)) # noqa: E731 + return get("pxf.Decimal"), get("pxf.BigFloat") + + +def _oracle_marshal(m) -> bytes: + """Go's oracleMarshal: proto.MarshalOptions{Deterministic: true}.""" + return m.SerializeToString(deterministic=True) + + +# --- The write half -------------------------------------------------------- + + +@pytest.mark.parametrize( + "value,want_scale", + [ + (decimal.Decimal("3.1415"), 4), + (decimal.Decimal("1.5"), 1), + (decimal.Decimal("0.001"), 3), + (decimal.Decimal("-2.25"), 2), + ], +) +def test_written_decimal_is_readable_by_schema(oracle, value, want_scale): + """Ports TestBignumConformance_WrittenBytesAreReadableBySchema, decimal/*. + + What this package emits must mean the same thing to the schema. 3.1415 + gives scale 4. Under zigzag that was written as 8, so a conformant reader + saw 31415 x 10^-8: the right digits, wrong by four orders of magnitude, + with nothing to signal it. + """ + decimal_cls, _ = oracle + data = pb.marshal(ConformanceStruct(price=value)) + o = decimal_cls() + # A schema-conformant decoder must accept what this package writes. + o.ParseFromString(_submessage(data, 1)) + # The scale a conformant reader sees must be the scale that was meant. + assert o.scale == want_scale + # And the value it reconstructs from those fields. + got = Fraction(int.from_bytes(o.unscaled, "big"), 10**o.scale) + if o.negative: + got = -got + assert got == Fraction(value), f"want {value}, oracle reconstructed {got}" + + +@pytest.mark.parametrize( + "text,prec", + [("1", 53), ("1.5", 53), ("1e10", 53), ("-1.23e-45", 64), ("6.02214076e23", 128)], + ids=["1.0", "1.5", "1e10", "-1.23e-45", "6.02214076e23"], +) +def test_written_bigfloat_is_readable_by_schema(oracle, text, prec): + """Ports TestBignumConformance_WrittenBytesAreReadableBySchema, bigfloat/*. + + BigFloat.exponent is adjExp = exp - prec, negative for essentially every + value, which is where protowire-go#92 bit hardest: 1.0 at prec 53 wrote an + exponent a conformant reader saw as 103 rather than -52, a factor of 2^155. + """ + _, bigfloat_cls = oracle + bf = _parse_float(text, prec) + data = pb.marshal(ConformanceStruct(coefficient=bf)) + o = bigfloat_cls() + o.ParseFromString(_submessage(data, 2)) + # value = mantissa x 2^exponent, per the schema, at the schema's prec. + got = _round_fraction(Fraction(int.from_bytes(o.mantissa, "big")) * Fraction(2) ** o.exponent, o.prec) + if o.negative: + got = pb.BigFloat(got.mantissa, got.exponent, got.prec, not got.negative) + assert got.to_fraction() == bf.to_fraction(), f"want {bf}, oracle reconstructed {got} (exponent {o.exponent})" + + +# --- The read half --------------------------------------------------------- + + +def test_schema_bytes_negative_scale(oracle): + """Ports TestBignumConformance_SchemaBytesAreReadableHere/negative scale. + + The read half, with a NEGATIVE scale written by the oracle. pb never + produces a negative scale itself, so this direction is the only way to + reach that arm. 25 x 10^-(-2) = 2500: a negative scale from a conformant + producer must mean trailing zeros. + """ + decimal_cls, _ = oracle + sub = _oracle_marshal(decimal_cls(unscaled=(25).to_bytes(1, "big"), scale=-2)) + got = pb.unmarshal(_wrap(1, sub), ConformanceStruct) + assert Fraction(got.price) == 2500, f"got {got.price}" + + +def test_schema_bytes_positive_scale(oracle): + """Ports TestBignumConformance_SchemaBytesAreReadableHere/positive scale. + + 31415 x 10^-4 = 3.1415. + """ + decimal_cls, _ = oracle + sub = _oracle_marshal(decimal_cls(unscaled=(31415).to_bytes(2, "big"), scale=4)) + got = pb.unmarshal(_wrap(1, sub), ConformanceStruct) + assert Fraction(got.price) == Fraction(31415, 10000), f"got {got.price}" + + +def test_schema_bytes_negative_exponent(oracle): + """Ports TestBignumConformance_SchemaBytesAreReadableHere/negative exponent. + + mantissa 2^52 x 2^-52 = 1.0, at prec 53. + """ + _, bigfloat_cls = oracle + mant = 1 << 52 + sub = _oracle_marshal(bigfloat_cls(mantissa=mant.to_bytes(7, "big"), exponent=-52, prec=53)) + got = pb.unmarshal(_wrap(2, sub), ConformanceStruct) + assert got.coefficient.to_fraction() == 1, f"got {got.coefficient}" + + +@pytest.mark.parametrize( + "wire_prec", [1 << 32, (1 << 32) + 1, (1 << 32) + 53, (1 << 33) + 64, (1 << 63) + 53, (1 << 64) - 1] +) +@pytest.mark.parametrize("neg", [False, True]) +def test_schema_bytes_prec_varint_past_uint32(oracle, wire_prec, neg): + """Ports TestBignumConformance_SchemaBytesAreReadableHere/prec varint past uint32. + + prec is a uint32, and a varint wider than 32 bits keeps its low 32 bits in + a protobuf reader. protowire-go once kept the whole varint and clamped it + to MaxPrec: 2^32 read as 2^32-1 bits there and as precision 0, a zero, to + the schema (protowire-go#230). + """ + _, bigfloat_cls = oracle + # 25 x 2^-3 at whatever precision the wire's low bits say. + sub = bytearray() + _append_tag(sub, 1, 2) + _append_bytes(sub, b"\x19") + _append_tag(sub, 2, 0) + _append_varint(sub, -3) + _append_tag(sub, 3, 0) + _append_varint(sub, wire_prec) + if neg: + _append_tag(sub, 4, 0) + _append_varint(sub, 1) + sub = bytes(sub) + + o = bigfloat_cls() + o.ParseFromString(sub) + prec = o.prec + assert prec == wire_prec & 0xFFFFFFFF, "the oracle keeps the low 32 bits" + + # The low 32 bits are what the prec bound sees too (protowire#499): + # 2^32+53 is 53 bits and reads, and only 2^64-1, 2^32-1 bits, is past + # the bound. (Checked before the expected value is built: Go's + # big.Float.SetPrec(2^32-1) costs nothing, but this test's exact model + # of the rounding would compute 2^(2^32-1).) + if prec > 13603: + with pytest.raises(ValueError, match="MaxNumericLiteralDigits"): + pb.unmarshal(_wrap(2, sub), ConformanceStruct) + return + + # The schema's value: precision 0 is a zero signed by negative, as in + # Unmarshal; otherwise mantissa x 2^exponent rounded to prec. + if prec == 0: + want = pb.BigFloat(0, 0, 0, neg) + else: + v = Fraction(int.from_bytes(o.mantissa, "big")) * Fraction(2) ** o.exponent + want = _round_fraction(-v if neg else v, prec) + got = pb.unmarshal(_wrap(2, sub), ConformanceStruct).coefficient + assert got.prec == prec, f"wire prec {wire_prec}" + assert got.to_fraction() == want.to_fraction(), f"wire prec {wire_prec}: got {got}, want {want}" + assert got.negative == neg, f"wire prec {wire_prec}" + + +def test_byte_for_byte(oracle): + """Ports TestBignumConformance_ByteForByte. + + Pins the encoding itself, so a future change to either side has to look + at the bytes rather than at a round trip that agrees with itself. + """ + decimal_cls, _ = oracle + ours = _submessage(pb.marshal(ConformanceStruct(price=decimal.Decimal("3.1415"))), 1) + theirs = _oracle_marshal(decimal_cls(unscaled=(31415).to_bytes(2, "big"), scale=4)) + # Field by field, not buffer by buffer: every field must encode to the + # same bytes, and the order they appear in is not part of the contract. + assert _field_records(ours) == _field_records(theirs) + # The specific byte the bug lived in: scale 4 is varint 0x04, not + # zigzag's 0x08. + assert b"\x10\x04" in ours, "field 2 varint must be 4 (plain), not 8 (zigzag)" diff --git a/tests/test_pb_conformance.py b/tests/test_pb_conformance.py new file mode 100644 index 0000000..104c26c --- /dev/null +++ b/tests/test_pb_conformance.py @@ -0,0 +1,297 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 TrendVidia, LLC. +"""protowire.pb against an independent oracle: a port of protowire-go's +encoding/pb/conformance_test.go. + +Tests that marshal with ``pb.marshal`` and read back with ``pb.unmarshal`` +pass for a codec that is wrong consistently. The oracle here is the protobuf +runtime over a schema compiled by protoc (testdata/pb/conf.proto, Go's +conformanceProtoSrc verbatim), and every case is checked both ways: + +- write: the bytes pb writes, read by the runtime, are the message the runtime + builds from the same values, and record by record they are the bytes the + runtime itself writes; +- read: the bytes the runtime writes decode here to the same values. + +The expected values are written in the schema's text format, never through +pb. As in Go, the type model has no fixed32/fixed64/sfixed*, no enum and no +oneof: the oneof cases set one member, and one test pins two. +""" + +from __future__ import annotations + +import math +import struct +from dataclasses import dataclass, field +from typing import Annotated, Optional + +import pytest +from google.protobuf import message_factory, text_format + +from protowire import pb + + +@dataclass +class ConfInner: + name: Annotated[str, pb.Field(1)] = "" + n: Annotated[int, pb.Field(2)] = 0 + + +@dataclass +class ConfAll: + s: Annotated[str, pb.Field(1)] = "" + i64: Annotated[int, pb.Field(2)] = 0 + i32: Annotated[int, pb.Field(3, pb.INT32)] = 0 + u64: Annotated[int, pb.Field(4, pb.UINT64)] = 0 + u32: Annotated[int, pb.Field(5, pb.UINT32)] = 0 + z64: Annotated[int, pb.Field(6, pb.SINT64)] = 0 + z32: Annotated[int, pb.Field(7, pb.SINT32)] = 0 + b: Annotated[bool, pb.Field(8)] = False + f64: Annotated[float, pb.Field(9)] = 0.0 + f32: Annotated[float, pb.Field(10, pb.FLOAT)] = 0.0 + buf: Annotated[bytes, pb.Field(11)] = b"" + inner: Annotated[Optional[ConfInner], pb.Field(12)] = None + lst: Annotated[list[ConfInner], pb.Field(13)] = field(default_factory=list) + nums: Annotated[list[int], pb.Field(14, pb.UINT32)] = field(default_factory=list) + zigs: Annotated[list[int], pb.Field(15, pb.SINT64)] = field(default_factory=list) + reals: Annotated[list[float], pb.Field(16, pb.FLOAT)] = field(default_factory=list) + strs: Annotated[list[str], pb.Field(17)] = field(default_factory=list) + m: Annotated[dict[str, int], pb.Field(18)] = field(default_factory=dict) + mm: Annotated[dict[Annotated[int, pb.INT32], Optional[ConfInner]], pb.Field(19)] = field(default_factory=dict) + a: Annotated[str, pb.Field(20)] = "" + c: Annotated[int, pb.Field(21)] = 0 + d: Annotated[Optional[ConfInner], pb.Field(22)] = None + flags: Annotated[list[bool], pb.Field(23)] = field(default_factory=list) + negs: Annotated[list[int], pb.Field(24)] = field(default_factory=list) + bm: Annotated[dict[bool, bool], pb.Field(25)] = field(default_factory=dict) + + +MAP_FIELDS = (18, 19, 25) +F32_MIN_SUBNORMAL = struct.unpack(" bytes: + return msg.SerializeToString(deterministic=True) + + +def _varint(b: bytes, i: int) -> tuple[int, int]: + v = shift = 0 + while True: + c = b[i] + i += 1 + v |= (c & 0x7F) << shift + if c < 0x80: + return v, i + shift += 7 + + +def records_by_field(b: bytes) -> dict[int, list[bytes]]: + """Top-level records (tag + value) by field number, in wire order. + + Map entries are sorted: their order is not part of the contract (the + runtime sorts by key under deterministic=True, pb keeps dict order). + """ + out: dict[int, list[bytes]] = {} + i = 0 + while i < len(b): + start = i + tag, i = _varint(b, i) + typ = tag & 7 + if typ == 0: + _, i = _varint(b, i) + elif typ == 1: + i += 8 + elif typ == 5: + i += 4 + elif typ == 2: + n, i = _varint(b, i) + i += n + else: + raise AssertionError(f"unexpected wire type {typ}") + out.setdefault(tag >> 3, []).append(b[start:i]) + for f in MAP_FIELDS: + if f in out: + out[f].sort() + return out + + +@pytest.mark.parametrize("name,ours,text", CASES, ids=IDS) +def test_write_direction(all_cls, name, ours, text): + """Ports TestConformance_WriteDirection.""" + want = oracle_message(all_cls, text) + data = pb.marshal(ours) + got = all_cls.FromString(data) # a conformant decoder reads what pb writes + assert got == want, f"oracle read our bytes as:\n{got}\nwant:\n{want}" + assert records_by_field(data) == records_by_field(oracle_marshal(want)), ( + "every record must be the bytes a conformant encoder writes" + ) + + +@pytest.mark.parametrize("name,ours,text", CASES, ids=IDS) +def test_read_direction(all_cls, name, ours, text): + """Ports TestConformance_ReadDirection.""" + theirs = oracle_marshal(oracle_message(all_cls, text)) + got = pb.unmarshal(theirs, ConfAll) + assert _same(got, ours) + + +def _same(a, b) -> bool: + """Equality that tells -0.0 from 0.0, as Go's assert.Equal does on floats.""" + if isinstance(a, float) and isinstance(b, float): + return struct.pack(" 0, f"{wrap_name}/{leaf_name}" + assert refused > 0, f"{wrap_name}/{leaf_name}" + + +def deep_node(n: int) -> DepthNode: + m = DepthNode(v=1) + for _ in range(n): + m = DepthNode(next=m) + return m + + +def test_depth_bound(): + """Ports TestMarshal_DepthBound.""" + out = pb.marshal(deep_node(pb.MAX_NESTING_DEPTH)) + pb.unmarshal(out, DepthNode()) + with pytest.raises(ValueError, match="nesting depth exceeds MaxNestingDepth=100"): + pb.marshal(deep_node(pb.MAX_NESTING_DEPTH + 1)) + + +def test_marshal_options_max_nesting_depth(): + """Ports TestMarshalOptions_MaxNestingDepth.""" + m = deep_node(150) + out = pb.marshal(m, max_nesting_depth=150) + pb.unmarshal(out, DepthNode(), max_nesting_depth=150) + with pytest.raises(ValueError, match="MaxNestingDepth=149"): + pb.marshal(m, max_nesting_depth=149) + with pytest.raises(ValueError, match="MaxNestingDepth=2"): + pb.marshal(deep_node(3), max_nesting_depth=2) + + +def test_encoder_depth(): + """Ports TestEncoder_Depth: a refused message writes no frame.""" + buf = io.BytesIO() + with pytest.raises(ValueError, match="MaxNestingDepth=100"): + pb.Encoder(buf).encode(deep_node(101)) + assert buf.getvalue() == b"", "a refused message writes no frame" + pb.Encoder(buf, max_nesting_depth=101).encode(deep_node(101)) + assert buf.getvalue() != b"" + + +def test_pointer_cycle(): + """Ports TestMarshal_PointerCycle: a cycle used to overflow Go's stack, which + no recover catches. Here it is refused by the depth bound, before + Python's recursion limit is reached.""" + m = DepthNode() + m.next = m + with pytest.raises(ValueError, match="nesting depth exceeds MaxNestingDepth=100"): + pb.marshal(m) diff --git a/tests/test_pb_limits.py b/tests/test_pb_limits.py new file mode 100644 index 0000000..599be96 --- /dev/null +++ b/tests/test_pb_limits.py @@ -0,0 +1,683 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 TrendVidia, LLC. +"""pb's decode limits (HARDENING.md § Mandatory limits), and the encoder's +refusal of what the decoder would reject. + +Ports protowire-go's encoding/pb limits_test.go, limits_options_test.go, +limits_size_test.go, internal/bigprec/bigprec_test.go, and the non-fuzzing +parts of fuzz_test.go. + +Go's ``promptly`` runs a decode in a goroutine and fails if it has not +returned within a budget, since a regression in a magnitude bound shows up as +work proportional to the hostile value (minutes to never). Here the call runs +inline and the elapsed time is asserted afterwards: a regression fails the +test late rather than promptly, but it still fails. +""" + +from __future__ import annotations + +import decimal +import io +import math +import time +from dataclasses import dataclass, field +from fractions import Fraction +from typing import Annotated, Optional + +import pytest + +from protowire import pb + +MAX_DIGITS = pb.MAX_NUMERIC_LITERAL_DIGITS +MAX_BFE = pb.MAX_BIG_FLOAT_EXPONENT +INT32_MIN, INT32_MAX = -(1 << 31), (1 << 31) - 1 +UINT32_MAX = (1 << 32) - 1 +BIG_MAX_PREC = UINT32_MAX # Go's big.MaxPrec + + +# --- Wire helpers (protobuf-go's protowire.Append*) ------------------------ + + +def _varint(out: bytearray, v: int) -> None: + v &= (1 << 64) - 1 # uint64(int64(x)): a negative is sign-extended + while v >= 0x80: + out.append((v & 0x7F) | 0x80) + v >>= 7 + out.append(v) + + +def _tag(out: bytearray, num: int, typ: int) -> None: + _varint(out, (num << 3) | typ) + + +def _blob(out: bytearray, b: bytes) -> None: + _varint(out, len(b)) + out += b + + +def _field1(sub: bytes) -> bytes: + out = bytearray() + _tag(out, 1, 2) + _blob(out, sub) + return bytes(out) + + +def decimal_wire(scale: int, negative: bool) -> bytes: + """Go's decimalWire: field 1 as a pxf.Decimal of 25 with the given scale, + by hand, so the decoder sees scales the encoder never writes.""" + sub = bytearray() + _tag(sub, 1, 2) + _blob(sub, bytes([25])) + _tag(sub, 2, 0) + # A plain int32 varint: sign-extended to 64 bits, as protobuf-go writes it. + _varint(sub, scale) + if negative: + _tag(sub, 3, 0) + _varint(sub, 1) + return _field1(bytes(sub)) + + +def bigfloat_wire(mant: bytes, exp: int, prec: int) -> bytes: + """Go's bigFloatWire: field 1 as a pxf.BigFloat, by hand.""" + sub = bytearray() + _tag(sub, 1, 2) + _blob(sub, mant) + _tag(sub, 2, 0) + _varint(sub, exp) + _tag(sub, 3, 0) + _varint(sub, prec) + return _field1(bytes(sub)) + + +def promptly(budget: float, f): + """Go's promptly, as a time check: call f, and fail if it took longer than + budget seconds. Returns f's exception, or None.""" + start = time.monotonic() + try: + f() + err = None + except ValueError as e: + err = e + elapsed = time.monotonic() - start + assert elapsed < budget, f"took {elapsed:.2f}s: the bound is not applied before the work it bounds" + return err + + +def mant_exp(f: pb.BigFloat) -> int: + """big.Float.MantExp(nil): the binary exponent, 0 for a zero.""" + return 0 if f.mantissa == 0 else f.exponent + f.mantissa.bit_length() + + +@dataclass +class DecimalHolder: + d: Annotated[Optional[decimal.Decimal], pb.Field(1)] = None + + +@dataclass +class BigFloatHolder: + f: Annotated[Optional[pb.BigFloat], pb.Field(1)] = None + + +@dataclass +class BigFloatList: + f: Annotated[list[pb.BigFloat], pb.Field(1)] = field(default_factory=list) + + +# --- limits_test.go -------------------------------------------------------- + + +@pytest.mark.parametrize( + "scale,want", + [ + (MAX_DIGITS, Fraction(25, 10**MAX_DIGITS)), + (-MAX_DIGITS, Fraction(25 * 10**MAX_DIGITS)), + (MAX_DIGITS + 1, None), + (-MAX_DIGITS - 1, None), + (INT32_MAX, None), + (INT32_MIN, None), + ], + ids=[ + "at the bound", + "at the negative bound", + "one past the bound", + "one past the negative bound", + "MaxInt32", + "MinInt32", + ], +) +def test_decimal_scale_bound(scale, want): + """Ports TestDecimalScaleBound (protowire-go#95). + + A Decimal whose scale magnitude exceeds MaxNumericLiteralDigits is + rejected before 10^|scale| is materialised, on both signs, and one at the + bound still decodes to its exact value. + """ + out = {} + err = promptly(5, lambda: out.setdefault("h", pb.unmarshal(decimal_wire(scale, False), DecimalHolder))) + if want is None: + assert err is not None, f"scale {scale}: decoded {out.get('h')}, want an error" + assert "MaxNumericLiteralDigits" in str(err), f"scale {scale}: error {err!r} does not name the limit" + return + assert err is None, f"scale {scale}: {err}" + assert Fraction(out["h"].d) == want + + +@pytest.mark.parametrize("exp", [-MAX_BFE - 5, -1, 0, 1, MAX_BFE - 5]) +def test_bigfloat_exponent_bound_inside(exp): + """Ports TestBigFloatExponentBound, the accepted half (protowire#281). + + Storing an exponent is cheap, but rendering the value in decimal costs + time that grows with the square of |exponent|, so the draft bounds it at + MaxBigFloatExponent. 25 has 5 bits, so the value's binary exponent is + exp + 5, and the bound is on that (MaxBigFloatExponent counts the value, + not the wire field). + """ + out = {} + err = promptly(5, lambda: out.setdefault("h", pb.unmarshal(bigfloat_wire(bytes([25]), exp, 64), BigFloatHolder))) + assert err is None, f"exponent {exp}: {err}" + # 25 = 0b11001, so v = 25 x 2^exp has MantExp exponent exp+5. + assert mant_exp(out["h"].f) == exp + 5 + + +@pytest.mark.parametrize("exp", [INT32_MIN, -MAX_BFE - 6, MAX_BFE - 4, INT32_MAX]) +def test_bigfloat_exponent_bound_past(exp): + """Ports TestBigFloatExponentBound, the refused half: past the bound, on + either side and up to the int32 extremes, decoding fails promptly.""" + err = promptly(5, lambda: pb.unmarshal(bigfloat_wire(bytes([25]), exp, 64), BigFloatHolder)) + assert err is not None and "MaxBigFloatExponent" in str(err), f"exponent {exp}: err = {err!r}" + + +def test_bigfloat_exponent_bound_per_call(): + """Ports TestBigFloatExponentBound: the bound is a per-call limit, as the + draft's limits are.""" + with pytest.raises(ValueError, match="MaxBigFloatExponent=10"): + pb.unmarshal(bigfloat_wire(bytes([25]), 6, 64), BigFloatHolder, max_big_float_exponent=10) + pb.unmarshal(bigfloat_wire(bytes([25]), 5, 64), BigFloatHolder, max_big_float_exponent=10) + + +# TestBigFloatExponentBoundEdges covers the states the bound's own test never +# reaches: a mantissa with no bits, with leading zero bytes, wider than its +# precision, a precision of zero, and a bound raised past big.Float's range. + + +@pytest.mark.parametrize("mant", [b"", b"\x00\x00"]) +@pytest.mark.parametrize("exp", [INT32_MIN, INT32_MAX]) +def test_bigfloat_edges_zero_mantissa(mant, exp): + """Ports TestBigFloatExponentBoundEdges/zero mantissa is zero at any exponent.""" + h = pb.unmarshal(bigfloat_wire(mant, exp, 64), BigFloatHolder) + assert h.f.mantissa == 0 + + +def test_bigfloat_edges_prec_zero(): + """Ports TestBigFloatExponentBoundEdges/prec 0 is zero at any exponent.""" + h = pb.unmarshal(bigfloat_wire(b"\x01", INT32_MIN, 0), BigFloatHolder) + assert h.f.mantissa == 0 + + +def test_bigfloat_edges_leading_zero_bytes(): + """Ports TestBigFloatExponentBoundEdges/leading zero bytes are not mantissa bits.""" + pb.unmarshal(bigfloat_wire(b"\x00\x00\x01", MAX_BFE - 1, 64), BigFloatHolder) + with pytest.raises(ValueError, match="MaxBigFloatExponent"): + pb.unmarshal(bigfloat_wire(b"\x00\x00\x01", MAX_BFE, 64), BigFloatHolder) + + +def test_bigfloat_edges_raised_bound_refuses_infinity(): + """Ports TestBigFloatExponentBoundEdges/a bound raised past big.Float's range + still refuses an infinity. + + protowire#278: 25 x 2^(2^31-1) is above big.Float's range. Go raises the + bound to math.MaxInt (2^63-1 on 64-bit), as here. + """ + with pytest.raises(ValueError, match="overflows"): + pb.unmarshal(bigfloat_wire(bytes([25]), INT32_MAX, 64), BigFloatHolder, max_big_float_exponent=(1 << 63) - 1) + + +def test_bigfloat_edges_rounding_carries_past_bound(): + """Ports TestBigFloatExponentBoundEdges/rounding to prec can carry past the bound. + + Pinned divergence. The bound is on the wire's mantissa and exponent, as + protowire#281 defines it; SetInt then rounds the mantissa to prec, and + 0xFF at one bit rounds up to 0x100. The decoded value is one past the + bound, and marshal refuses it. This fails the day the decoder checks the + rounded value instead. + """ + h = pb.unmarshal(bigfloat_wire(b"\xff", MAX_BFE - 8, 1), BigFloatHolder) + assert mant_exp(h.f) == MAX_BFE + 1 + with pytest.raises(ValueError, match="MaxBigFloatExponent"): + pb.marshal(h) + + +def test_bigfloat_prec_bound(): + """Ports TestBigFloatPrecBound (protowire#499). + + A BigFloat's prec is bounded as a precision suffix is, by the digits N its + literal needs: 13603 bits at the default reads, 13604 does not. Marshal + writes the mantissa widened to prec bits, so an 11-byte message at 2^31-1 + re-encoded to 256 MiB; it is refused before anything is built for it. + + Go also asserts, with runtime.MemStats, that each refusal allocates under + 64 KiB (a widened mantissa at 2^27 bits is 16 MiB). That check is replaced + by a time bound here: a refusal that built the widened mantissa first + would take far longer than the budget at 2^31-1 bits. + """ + h = pb.unmarshal(bigfloat_wire(b"\x19", -3, 13603), BigFloatHolder) + assert h.f.prec == 13603 and h.f.to_fraction() == Fraction(3125, 1000) + for prec in [13604, 1 << 27, INT32_MAX, UINT32_MAX]: + err = promptly(0.5, lambda: pb.unmarshal(bigfloat_wire(b"\x19", -3, prec), BigFloatHolder)) + assert err is not None and "MaxNumericLiteralDigits=4096" in str(err), f"prec {prec}: err = {err!r}" + # The bound is a per-call limit, through MaxNumericLiteralDigits: 2^29 + # has 9 digits, so N is 10, and 2^30 has 10. + pb.unmarshal(bigfloat_wire(b"\x19", -3, 29), BigFloatHolder, max_numeric_literal_digits=10) + with pytest.raises(ValueError, match="MaxNumericLiteralDigits=10"): + pb.unmarshal(bigfloat_wire(b"\x19", -3, 30), BigFloatHolder, max_numeric_literal_digits=10) + pb.unmarshal(bigfloat_wire(b"\x19", -3, 13604), BigFloatHolder, max_numeric_literal_digits=5000) + + +# TestMarshalRefusesWhatUnmarshalRejects: the encoders refuse a value the wire +# cannot carry, instead of writing bytes no conformant decoder accepts (a +# scale past the limit), a wrong exponent (int32 wrap), or panicking (an +# infinity). + + +def _one_over_two_pow(k: int) -> decimal.Decimal: + """1/2^k exactly: 5^k x 10^-k, built from a string so no context rounds.""" + return decimal.Decimal(f"{5**k}E-{k}") + + +def test_marshal_refuses_decimal_scale_past_limit(): + """Ports TestMarshalRefusesWhatUnmarshalRejects/Decimal scale past the limit.""" + with pytest.raises(ValueError, match="MaxNumericLiteralDigits"): + pb.marshal(DecimalHolder(d=_one_over_two_pow(MAX_DIGITS + 1))) + + +def test_marshal_decimal_scale_at_limit_round_trips(): + """Ports TestMarshalRefusesWhatUnmarshalRejects/Decimal scale at the limit round-trips.""" + v = _one_over_two_pow(MAX_DIGITS) + assert Fraction(v) == Fraction(1, 2**MAX_DIGITS) + out = pb.unmarshal(pb.marshal(DecimalHolder(d=v)), DecimalHolder) + assert Fraction(out.d) == Fraction(v), "round trip changed the value" + + +# TestMarshalRefusesWhatUnmarshalRejects/infinite BigFloat is not ported: +# pb.BigFloat has no infinity. A big.Float is ±Inf after an overflow; a +# pb.BigFloat whose binary exponent passes int32 is refused at construction +# instead (test_pb_bigfloat.py::test_overflow_above_int32_is_an_error), so the +# value Go's test marshals cannot be built here. + + +@pytest.mark.parametrize("prec", [1 << 31, (1 << 31) + 1, BIG_MAX_PREC, (1 << 32) + (1 << 31)]) +def test_bigfloat_prec_too_wide_unmarshal(prec): + """Ports TestMarshalRefusesWhatUnmarshalRejects/BigFloat precision too wide to + scale, the decode half. + + protowire-go#229's eleven bytes: Unmarshal stored the precision, and + Marshal used to dereference nil. Unmarshal now refuses the precision + itself (protowire#499). A varint past uint32 keeps its low 32 bits + (protowire-go#230), so 1<<32 + 1<<31 is 1<<31 again. + """ + sub = bytearray() + _tag(sub, 1, 2) + _blob(sub, b"\x19") + _tag(sub, 3, 0) + _varint(sub, prec) + wire = _field1(bytes(sub)) + if prec == 1 << 31: + assert wire == bytes([0x0A, 0x09, 0x0A, 0x01, 0x19, 0x18, 0x80, 0x80, 0x80, 0x80, 0x08]), "not the reported message" + with pytest.raises(ValueError, match="MaxNumericLiteralDigits"): + pb.unmarshal(wire, BigFloatHolder(f=pb.BigFloat())) + + +@pytest.mark.parametrize("prec", [1 << 31, 13604]) +def test_bigfloat_prec_too_wide_marshal(prec): + """Ports TestMarshalRefusesWhatUnmarshalRejects/BigFloat precision too wide to + scale, the encode half: the same value built in memory, on both signs and + through a list, and the one past the bound. Marshal refuses what Unmarshal + would.""" + want = "MaxNumericLiteralDigits=4096" + neg = pb.BigFloat(3, 0, prec, True) # SetPrec(prec).SetInt64(-3) + with pytest.raises(ValueError, match=want): + pb.marshal(BigFloatHolder(f=neg)) + with pytest.raises(ValueError, match=want): + pb.marshal(BigFloatList(f=[neg])) + + +def test_bigfloat_prec_at_bound_round_trips(): + """Ports TestMarshalRefusesWhatUnmarshalRejects/BigFloat precision too wide to + scale: at the bound the value is written and reads back.""" + at = pb.BigFloat(3, 0, 13603, True) + back = pb.unmarshal(pb.marshal(BigFloatHolder(f=at)), BigFloatHolder) + assert back.f.to_fraction() == at.to_fraction() and back.f.prec == 13603 + + +def test_marshal_bigfloat_exponent_below_int32_after_adjustment(): + """Ports TestMarshalRefusesWhatUnmarshalRejects/BigFloat exponent below int32 + after adjustment. + + exp - prec leaves int32: 0.75 x 2^(MinInt32+3) with 53 bits of precision + has wire exponent MinInt32+3-53. (0.75 x 2^e is 3 x 2^(e-2).) + """ + f = pb.BigFloat(3, INT32_MIN + 3 - 2, 53) + with pytest.raises(ValueError, match="does not fit int32"): + pb.marshal(BigFloatHolder(f=f)) + + +def test_marshal_bigfloat_exponent_past_max(): + """Ports TestMarshalRefusesWhatUnmarshalRejects/BigFloat exponent past + MaxBigFloatExponent. 0.75 x 2^k has binary exponent k.""" + f = pb.BigFloat(3, -MAX_BFE - 1 - 2, 53) + assert mant_exp(f) == -MAX_BFE - 1 + with pytest.raises(ValueError, match="MaxBigFloatExponent"): + pb.marshal(BigFloatHolder(f=f)) + + +def test_marshal_bigfloat_exponent_at_max_round_trips(): + """Ports TestMarshalRefusesWhatUnmarshalRejects/BigFloat exponent at + MaxBigFloatExponent round-trips.""" + f = pb.BigFloat(3, -MAX_BFE - 2, 53) + out = pb.unmarshal(pb.marshal(BigFloatHolder(f=f)), BigFloatHolder(f=pb.BigFloat())) + assert out.f.to_fraction() == f.to_fraction(), f"round trip changed the value: {f} -> {out.f}" + + +# --- limits_options_test.go ------------------------------------------------ + + +@dataclass +class NestHolder: + """Go's nestHolder: nests itself, so nest_to(n) is n structs deep.""" + + child: Annotated[Optional[NestHolder], pb.Field(1)] = None + v: Annotated[int, pb.Field(2, pb.INT32)] = 0 + tags: Annotated[dict[str, str], pb.Field(3)] = field(default_factory=dict) + + +def nest_to(depth: int) -> NestHolder: + root = NestHolder(v=1) + cur = root + for i in range(1, depth): + cur.child = NestHolder(v=i + 1) + cur = cur.child + return root + + +def test_unmarshal_options_max_nesting_depth(): + """Ports TestUnmarshalOptionsLimits/MaxNestingDepth (protowire-go#101). + + The per-call limits default to the package constants at 0, and a lowered + value rejects input the default accepts. + """ + data = pb.marshal(nest_to(3)) + pb.unmarshal(data, NestHolder()) # default + pb.unmarshal(data, NestHolder(), max_nesting_depth=0) # zero options are the default + # nest_to(3) is the root and two descents; the root is depth 0. + pb.unmarshal(data, NestHolder(), max_nesting_depth=2) # at the bound + with pytest.raises(ValueError, match="MaxNestingDepth=1") as ei: + pb.unmarshal(data, NestHolder(), max_nesting_depth=1) + # The error names the effective limit, not the constant. + assert "MaxNestingDepth=100" not in str(ei.value) + + +def test_unmarshal_options_max_numeric_literal_digits(): + """Ports TestUnmarshalOptionsLimits/MaxNumericLiteralDigits.""" + wire = decimal_wire(10, False) + pb.unmarshal(wire, DecimalHolder) # default + pb.unmarshal(wire, DecimalHolder, max_numeric_literal_digits=10) # at the bound + with pytest.raises(ValueError, match="MaxNumericLiteralDigits=9"): + pb.unmarshal(wire, DecimalHolder, max_numeric_literal_digits=9) + # Both signs of the scale are bounded by the call's value. + with pytest.raises(ValueError, match="MaxNumericLiteralDigits=9"): + pb.unmarshal(decimal_wire(-10, False), DecimalHolder, max_numeric_literal_digits=9) + + +def test_unmarshal_options_stream_decoder(): + """Ports TestUnmarshalOptionsLimits/stream decoder: the length-prefixed stream + decoder takes the same limits.""" + buf = io.BytesIO() + pb.Encoder(buf).encode(nest_to(3)) + framed = buf.getvalue() + assert pb.Decoder(io.BytesIO(framed)).decode(NestHolder()) is not None # default + assert pb.Decoder(io.BytesIO(framed), max_nesting_depth=0).decode(NestHolder()) is not None # zero options + with pytest.raises(ValueError, match="MaxNestingDepth=1"): + pb.Decoder(io.BytesIO(framed), max_nesting_depth=1).decode(NestHolder()) + + +@pytest.mark.parametrize("depth,ok", [(pb.MAX_NESTING_DEPTH + 1, True), (pb.MAX_NESTING_DEPTH + 2, False)]) +def test_nesting_depth_boundary(depth, ok): + """Ports TestNestingDepthBoundary (protowire-go#111). + + The root is depth 0 and every submessage is one descent (HARDENING.md + § Recursion), so MaxNestingDepth submessages under the root (the corpus's + deep-submessage-100) is the last accepted shape. + """ + data = pb.marshal(nest_to(depth), max_nesting_depth=1000) + if ok: + pb.unmarshal(data, NestHolder()) + else: + with pytest.raises(ValueError, match="MaxNestingDepth=100"): + pb.unmarshal(data, NestHolder()) + + +@pytest.mark.parametrize("children,ok", [(pb.MAX_NESTING_DEPTH - 1, True), (pb.MAX_NESTING_DEPTH, False)]) +def test_nesting_depth_boundary_scalar_map_entry(children, ok): + """Ports TestNestingDepthBoundaryScalarMapEntry (protowire#449). + + A map entry is a descent whatever its value's type, so a scalar-valued + entry just past the bound is refused. + """ + msg = NestHolder(tags={"k": "v"}) + for _ in range(children): + msg = NestHolder(child=msg) + data = pb.marshal(msg, max_nesting_depth=1000) + if ok: + pb.unmarshal(data, NestHolder()) + else: + with pytest.raises(ValueError, match="MaxNestingDepth=100"): + pb.unmarshal(data, NestHolder()) + + +# --- limits_size_test.go --------------------------------------------------- + + +@dataclass +class BlobHolder: + b: Annotated[bytes, pb.Field(1)] = b"" + + +@dataclass +class ListHolder: + xs: Annotated[list[int], pb.Field(1, pb.INT32)] = field(default_factory=list) + m: Annotated[dict[str, int], pb.Field(2, pb.INT32)] = field(default_factory=dict) + ss: Annotated[list[str], pb.Field(3)] = field(default_factory=list) + + +def test_max_message_size_default_rejects(): + """Ports TestMaxMessageSize/default rejects 64 MiB + tag (protowire-go#112). + + The input to one decode call is capped at MaxMessageSize before anything + is read. + """ + data = pb.marshal(BlobHolder(b=bytes(pb.MAX_MESSAGE_SIZE))) + assert len(data) > pb.MAX_MESSAGE_SIZE, "tag and length push it past the cap" + with pytest.raises(ValueError, match="MaxMessageSize=67108864"): + pb.unmarshal(data, BlobHolder) + + +def test_max_message_size_default_accepts(): + """Ports TestMaxMessageSize/default accepts 64 MiB.""" + data = pb.marshal(BlobHolder(b=bytes(pb.MAX_MESSAGE_SIZE - 16))) + assert len(data) <= pb.MAX_MESSAGE_SIZE + pb.unmarshal(data, BlobHolder) + + +def test_max_message_size_lowered_per_call(): + """Ports TestMaxMessageSize/lowered per call, on unmarshal and on the stream + decoder alike.""" + data = pb.marshal(BlobHolder(b=bytes(32))) + pb.unmarshal(data, BlobHolder) # default + pb.unmarshal(data, BlobHolder, max_message_size=len(data)) # at the bound + with pytest.raises(ValueError, match="MaxMessageSize=16"): + pb.unmarshal(data, BlobHolder, max_message_size=16) + + # A frame inside the stream's frame cap, payload past the call's cap. + buf = io.BytesIO() + pb.Encoder(buf).encode(BlobHolder(b=bytes(32))) + assert pb.Decoder(io.BytesIO(buf.getvalue())).decode(BlobHolder) is not None # default + with pytest.raises(ValueError, match="MaxMessageSize=16"): + pb.Decoder(io.BytesIO(buf.getvalue()), max_message_size=16).decode(BlobHolder) + + +def test_max_repeated_count(): + """Ports TestMaxRepeatedCount (protowire-go#112). + + The element count of a repeated or map field is capped at + MaxRepeatedCount, lowered per call, on packed and unpacked elements and + on map entries. + """ + data = pb.marshal(ListHolder(xs=[1, 2, 3, 4, 5], m={"a": 1, "b": 2, "c": 3}, ss=["a", "b", "c"])) + pb.unmarshal(data, ListHolder) # default + pb.unmarshal(data, ListHolder, max_repeated_count=5) # at the bound + with pytest.raises(ValueError, match="MaxRepeatedCount=4"): # five packed int32s + pb.unmarshal(data, ListHolder, max_repeated_count=4) + with pytest.raises(ValueError, match="MaxRepeatedCount=2"): + pb.unmarshal(data, ListHolder, max_repeated_count=2) + + # Unpacked strings and map entries hit the same bound. + data = pb.marshal(ListHolder(ss=["a", "b", "c"])) + with pytest.raises(ValueError, match="MaxRepeatedCount=2"): + pb.unmarshal(data, ListHolder, max_repeated_count=2) + data = pb.marshal(ListHolder(m={"a": 1, "b": 2, "c": 3})) + with pytest.raises(ValueError, match="MaxRepeatedCount=2"): + pb.unmarshal(data, ListHolder, max_repeated_count=2) + pb.unmarshal(data, ListHolder, max_repeated_count=3) # three entries at the bound + + +# --- internal/bigprec/bigprec_test.go -------------------------------------- + + +def test_pow2_digits(): + """Ports bigprec's TestPow2Digits. + + _pow2_digits counts the decimal digits of 2^p without computing 2^p. It is + checked against the count of the power itself for every p up to 2^15, + which takes in the default limit's bound, p13603, with room. The count is + kept as Go keeps it, by tracking the smallest power of ten above 2^p, + rather than as len(str(2**p)): that is the same number, and str() past + 4300 digits is refused by Python's int-to-str limit and quadratic below + it. The p <= 2^12 prefix is also checked with len(str(2**p)) directly. + + Past that the exact counts are pinned from an independent computation + (log10(2) to 2000 bits, by series), at the three p below 2^32 where + float64's p*log10(2) is one digit high, which is why the helper does not + use it. + """ + pow2, nxt, want = 1, 10, 1 + for p in range(0, (1 << 15) + 1): + if p > 0: + pow2 <<= 1 + if pow2 >= nxt: + want += 1 + nxt *= 10 + assert pb._pow2_digits(p) == want, f"_pow2_digits({p})" + if p <= 1 << 12: + assert want == len(str(pow2)), f"p {p}: the incremental count disagrees with str()" + for p, want in [(146964308, 44240665), (345060773, 103873643), (1923400330, 579001193)]: + assert pb._pow2_digits(p) == want, f"_pow2_digits({p})" + f = int(p * math.log10(2)) + 1 + assert f != want, f"p {p}: float64 gives {f} too; the reason for fixed point no longer holds here" + # The draft's form of the bound: N exceeds L exactly when prec exceeds + # floor((L-1)/log10(2)), 13603 at the default of 4096. (Go's Digits(prec, + # limit) is (_prec_digits(prec), _prec_digits(prec) <= limit).) + assert pb._prec_digits(13603) <= 4096, "p13603 is within the default limit" + assert pb._prec_digits(13604) > 4096, "p13604 is past the default limit" + assert pb._prec_digits(UINT32_MAX) == 1292913988 + + +# --- fuzz_test.go ---------------------------------------------------------- + + +@dataclass +class FuzzInner: + name: Annotated[str, pb.Field(1)] = "" + n: Annotated[int, pb.Field(2, pb.INT32)] = 0 + + +@dataclass +class FuzzTarget: + """Go's fuzzTarget: scalars, nested message, repeated message, byte slice, + zigzag.""" + + s: Annotated[str, pb.Field(1)] = "" + i: Annotated[int, pb.Field(2)] = 0 + u: Annotated[int, pb.Field(3, pb.UINT32)] = 0 + b: Annotated[bool, pb.Field(4)] = False + f32: Annotated[float, pb.Field(5, pb.FLOAT)] = 0.0 + f64: Annotated[float, pb.Field(6)] = 0.0 + z: Annotated[int, pb.Field(7, pb.SINT64)] = 0 + buf: Annotated[bytes, pb.Field(8)] = b"" + inner: Annotated[Optional[FuzzInner], pb.Field(9)] = None + list_: Annotated[list[Optional[FuzzInner]], pb.Field(10)] = field(default_factory=list) + nums: Annotated[list[int], pb.Field(11, pb.UINT32)] = field(default_factory=list) # packed repeated scalar + zigs: Annotated[list[int], pb.Field(12, pb.SINT64)] = field(default_factory=list) + reals: Annotated[list[float], pb.Field(13, pb.FLOAT)] = field(default_factory=list) + rat: Annotated[Optional[decimal.Decimal], pb.Field(14)] = None + bf: Annotated[Optional[pb.BigFloat], pb.Field(15)] = None + + +def hostile_decimal(scale: int) -> bytes: + """Go's hostileDecimal: field 14 as a Decimal of 25 at the given scale.""" + sub = bytearray() + _tag(sub, 1, 2) + _blob(sub, bytes([25])) + _tag(sub, 2, 0) + _varint(sub, scale) + out = bytearray() + _tag(out, 14, 2) + _blob(out, bytes(sub)) + return bytes(out) + + +@pytest.mark.parametrize("scale", [INT32_MAX, INT32_MIN]) +def test_hostile_decimal_seeds_reach_the_bound(scale): + """Ports TestHostileDecimalSeedsReachTheBound. + + The positive marker that the hostile seeds exercise the arm they are for: + each must fail on the limit, not on a corrupt tag or a silent skip. + """ + with pytest.raises(ValueError, match="MaxNumericLiteralDigits"): + pb.unmarshal(hostile_decimal(scale), FuzzTarget) + + +_FUZZ_SEEDS = [ + FuzzTarget(s="hello", i=42, u=7, b=True, f32=1.5, f64=2.5), + FuzzTarget(z=-100, buf=bytes([0x00, 0xFF, 0x10])), + FuzzTarget(inner=FuzzInner(name="x", n=-1)), + FuzzTarget(list_=[FuzzInner(name="a", n=1), FuzzInner(name="b", n=2)]), + FuzzTarget(nums=[1, 0, 300], zigs=[-1, 0, 2], reals=[1.5, 0]), + FuzzTarget(), +] + + +def _fuzz_corpus() -> list[bytes]: + corpus = [pb.marshal(s) for s in _FUZZ_SEEDS] + corpus += [ + b"", + b"\xff", + b"\x08", # tag-only, truncated varint + bytes([0x0A, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x7F]), # length-delim with absurd length + # A Decimal whose scale is 2^31-1 and one at -2^31, so the corpus + # reaches the bound (protowire-go#95). + hostile_decimal(INT32_MAX), + hostile_decimal(INT32_MIN), + ] + return corpus + + +@pytest.mark.parametrize("data", _fuzz_corpus()) +def test_unmarshal_seed_corpus(data): + """Ports FuzzUnmarshal's seed corpus, which `go test` runs without -fuzz. + + Any out-of-bounds read, infinite loop or crash is a bug: unmarshal must + return or raise ValueError on malformed input. The fuzzing itself (Go's + coverage-guided mutation of these seeds) is not ported: Python has no + equivalent in this suite. + """ + promptly(5, lambda: pb.unmarshal(data, FuzzTarget)) diff --git a/tests/test_pb_stream.py b/tests/test_pb_stream.py new file mode 100644 index 0000000..865af17 --- /dev/null +++ b/tests/test_pb_stream.py @@ -0,0 +1,139 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 TrendVidia, LLC. +"""pb.Encoder / pb.Decoder against protowire-go's encoding/pb/stream_test.go (v1.17.0). + +A frame is a varint length and then the message. Go's Decoder returns io.EOF +at a clean end of stream, io.ErrUnexpectedEOF on a truncated frame and +ErrFramingCorrupt on bad framing. Here they are ``None``, ``EOFError`` and +``pb.FramingError``. The last two are sticky, as in Go. + +Not ported: + +- TestStreamScratchReuse checks that Go's Decoder reuses its internal + scratch buffer (``&dec.buf[0]``) between same-size frames. That is an + allocation detail of the Go implementation with no observable behaviour; + the Python Decoder has no such buffer. +- FuzzStreamDecode is a Go fuzz target, not a test case. +""" + +from __future__ import annotations + +import io +from dataclasses import dataclass +from typing import Annotated + +import pytest + +from protowire import pb + + +@dataclass +class StreamMsg: + n: Annotated[int, pb.Field(1)] = 0 + text: Annotated[str, pb.Field(2)] = "" + data: Annotated[bytes, pb.Field(3)] = b"" + + +def put_uvarint(v: int) -> bytes: + """encoding/binary.PutUvarint.""" + out = bytearray() + while v >= 0x80: + out.append((v & 0x7F) | 0x80) + v >>= 7 + out.append(v) + return bytes(out) + + +def test_stream_round_trip(): + """Ports TestStreamRoundTrip.""" + msgs = [ + StreamMsg(n=1, text="small"), + StreamMsg(n=2, text="medium " * 128), + StreamMsg(n=3, data=b"\xab" * (1 << 20)), + ] + buf = io.BytesIO() + enc = pb.Encoder(buf) + for m in msgs: + enc.encode(m) + + buf.seek(0) + dec = pb.Decoder(buf) + for i, want in enumerate(msgs): + assert dec.decode(StreamMsg()) == want, f"frame {i} mismatch" + assert dec.decode(StreamMsg()) is None # Go: io.EOF + + +def test_stream_clean_eof(): + """Ports TestStreamCleanEOF.""" + assert pb.Decoder(io.BytesIO()).decode(StreamMsg()) is None # Go: io.EOF + + +def test_stream_partial_prefix(): + """Ports TestStreamPartialPrefix.""" + dec = pb.Decoder(io.BytesIO(bytes([0x80]))) + with pytest.raises(EOFError) as e1: # Go: io.ErrUnexpectedEOF + dec.decode(StreamMsg()) + with pytest.raises(EOFError) as e2: + dec.decode(StreamMsg()) + assert e2.value is e1.value, "not sticky" + + +def test_stream_partial_body(): + """Ports TestStreamPartialBody.""" + buf = io.BytesIO(put_uvarint(100) + bytes(10)) + dec = pb.Decoder(buf) + with pytest.raises(EOFError) as e1: # Go: io.ErrUnexpectedEOF + dec.decode(StreamMsg()) + with pytest.raises(EOFError) as e2: + dec.decode(StreamMsg()) + assert e2.value is e1.value, "not sticky" + + +def test_stream_oversized(): + """Ports TestStreamOversized.""" + buf = io.BytesIO(put_uvarint(1 << 30)) + dec = pb.Decoder(buf, max_frame_size=1 << 20) # Go: dec.SetMaxFrameSize(1 << 20) + with pytest.raises(pb.FramingError): + dec.decode(StreamMsg()) + with pytest.raises(pb.FramingError): # sticky + dec.decode(StreamMsg()) + + +def test_stream_varint_overflow(): + """Ports TestStreamVarintOverflow.""" + dec = pb.Decoder(io.BytesIO(b"\xff" * 11)) + with pytest.raises(pb.FramingError): + dec.decode(StreamMsg()) + + +def test_stream_inner_unmarshal_recoverable(): + """Ports TestStreamInnerUnmarshalRecoverable.""" + # A frame with a corrupt body shouldn't desync the framing — the next + # frame is at a known boundary, so the caller can keep going. + good = StreamMsg(n=7, text="ok") + buf = io.BytesIO() + enc = pb.Encoder(buf) + enc.encode(good) + buf.write(put_uvarint(4)) + buf.write(b"\xff\xff\xff\xff") + enc.encode(good) + + buf.seek(0) + dec = pb.Decoder(buf) + assert dec.decode(StreamMsg()) == good + with pytest.raises(ValueError) as e: + dec.decode(StreamMsg()) + assert not isinstance(e.value, pb.FramingError), f"got framing error, expected inner unmarshal: {e.value}" + got3 = dec.decode(StreamMsg()) + assert got3 is not None and got3.n == 7 and got3.text == "ok" + + +@pytest.mark.parametrize("size", [0, 1, 127, 128, 16383, 16384, 1 << 17]) +def test_stream_length_boundaries(size): + """Ports TestStreamLengthBoundaries.""" + buf = io.BytesIO() + msg = StreamMsg(data=b"\xcc" * size) + pb.Encoder(buf).encode(msg) + buf.seek(0) + got = pb.Decoder(buf).decode(StreamMsg()) + assert got is not None and got.data == msg.data From 30fb3a6ab122aa5f08192b5ab61a7de71ddb7f1f Mon Sep 17 00:00:00 2001 From: "Benedicto Franco Jr." Date: Sat, 7 Nov 2026 18:53:20 -0800 Subject: [PATCH 2/5] check_decode: decode --format pb through protowire.pb As Go's harness decodes through its encoding/pb, into dataclasses that mirror adversarial.proto. All 29 pb corpus rows pass, with the same verdict as protowire-go's harness on each. --- scripts/check_decode.py | 77 ++++++++++++++++++++++++++++++++++------- 1 file changed, 65 insertions(+), 12 deletions(-) diff --git a/scripts/check_decode.py b/scripts/check_decode.py index 7211a86..b9a9ad3 100644 --- a/scripts/check_decode.py +++ b/scripts/check_decode.py @@ -34,9 +34,9 @@ - pxf and sbe go through ``protowire.pxf`` and ``protowire.sbe``, that is through the protowire-cpp decoder this port wraps. -- pb goes through the upstream protobuf runtime. This port has no PB decoder - of its own, so these rows test what a caller gets from - ``Message.ParseFromString``, and a fix for one of them cannot land here. +- pb goes through ``protowire.pb``, this port's own PB codec, as the Go + reference's goes through its encoding/pb: the bytes decode into the + hand-declared dataclasses below, which mirror adversarial.proto. - pb-render writes the bytes as PXF through ``pxf.marshal_bytes``, so these rows test the protowire-cpp encoder. The bridge reads the bytes one level deeper than the encoder's limit, leaving the encoder to refuse. A read that @@ -62,10 +62,10 @@ ``--limit MaxMessageSize=1024`` sets one HARDENING limit for the row, under the Go name the manifest uses. pxf rows take all six, sbe rows MaxMessageSize and MaxRepeatedCount, and pb-render MaxNestingDepth. pxf-pb -takes the six, as pxf does, and pb-text none, as in the Go reference. A limit -this port cannot apply to the row's format exits 2: running the row without -it would not test the limit. pb rows take none, since the protobuf runtime -has no such settings. +takes the six, as pxf does, and pb-text none, as in the Go reference. pb takes +the five protowire.pb has: all but MaxBytesLiteralLength, which bounds a PXF +literal. A limit this port cannot apply to the row's format exits 2: running +the row without it would not test the limit. """ from __future__ import annotations @@ -73,13 +73,16 @@ import argparse import struct import sys +from dataclasses import dataclass, field +from decimal import Decimal from pathlib import Path +from typing import Annotated, Optional try: from google.protobuf import descriptor_pb2, descriptor_pool, message_factory from google.protobuf.message import DecodeError - from protowire import pxf, sbe + from protowire import pb, pxf, sbe except ImportError as e: # A broken build must not read as a reject of every row. print(f"check_decode: {e}", file=sys.stderr) sys.exit(2) @@ -117,7 +120,7 @@ def _message_class(fds_bytes: bytes, schema: str): "pxf-pb": set(_LIMITS.values()), "sbe": {"max_message_size", "max_repeated_count"}, "pb-render": {"max_nesting_depth"}, - "pb": set(), + "pb": set(_LIMITS.values()) - {"max_bytes_literal_length"}, "pb-text": set(), "envelope": set(), } @@ -254,11 +257,61 @@ def _pb_text(fds_bytes: bytes, schema: str, data: bytes, limits: dict[str, int]) return 0 +# Hand-declared mirrors of adversarial.proto for protowire.pb, which reads +# dataclasses rather than descriptors. Drift from the .proto shows up as a +# corpus row failing its expectation. Unlike the Go reference's mirror, Tree +# keeps `children`, so the mirror is the whole message. + + +@dataclass +class Tree: + child: Annotated[Optional[Tree], pb.Field(1)] = None + label: Annotated[str, pb.Field(2)] = "" + children: Annotated[list[Tree], pb.Field(3)] = field(default_factory=list) + branches: Annotated[dict[str, Optional[Tree]], pb.Field(4)] = field(default_factory=dict) + tags: Annotated[dict[str, str], pb.Field(5)] = field(default_factory=dict) + + +@dataclass +class StringHolder: + value: Annotated[str, pb.Field(1)] = "" + + +@dataclass +class BytesHolder: + value: Annotated[bytes, pb.Field(1)] = b"" + + +@dataclass +class BigIntHolder: + value: Annotated[int, pb.Field(1)] = 0 + + +@dataclass +class ListHolder: + values: Annotated[list[int], pb.Field(1, pb.INT32)] = field(default_factory=list) + + +@dataclass +class BigNumHolder: + big_int: Annotated[Optional[int], pb.Field(1, pb.BIGINT)] = None + decimal: Annotated[Optional[Decimal], pb.Field(2)] = None + big_float: Annotated[Optional[pb.BigFloat], pb.Field(3)] = None + + +_PB_TYPES = { + "adversarial.v1." + c.__name__: c + for c in (Tree, StringHolder, BytesHolder, BigIntHolder, ListHolder, BigNumHolder) +} + + def _decode_pb(fds_bytes: bytes, schema: str, data: bytes, limits: dict[str, int]) -> int: - msg = _message_class(fds_bytes, schema)() + cls = _PB_TYPES.get(schema) + if cls is None: + raise SystemExit(f"check_decode: no protowire.pb mirror for {schema}") try: - msg.ParseFromString(data) - except DecodeError as e: + pb.unmarshal(data, cls, **limits) + except ValueError as e: return _reject(f"pb: {e}") return 0 From ec3b223ebbeca988b50d2e32eb4da419169a6c8c Mon Sep 17 00:00:00 2001 From: "Benedicto Franco Jr." Date: Sat, 7 Nov 2026 18:53:20 -0800 Subject: [PATCH 3/5] docs: document protowire.pb --- CHANGELOG.md | 20 ++++++++++++++++++++ README.md | 41 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f73c12..475287b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,26 @@ format changes. ## [Unreleased] +### Added + +- **`protowire.pb`, a PB codec for dataclasses** (#36, #34): a port of + protowire-go's `encoding/pb`. Fields are declared with + `Annotated[T, pb.Field(number, kind)]`; `pb.marshal`, `pb.unmarshal` and the + length-prefixed `pb.Encoder` / `pb.Decoder` write and read protowire-go's + bytes, big numbers included (`decimal.Decimal` for `pxf.Decimal`, + `pb.BigFloat` for `pxf.BigFloat`, `pb.BIGINT` for `pxf.BigInt`). Decoding + takes Go's per-call limits as keyword arguments (`max_message_size`, + `max_nesting_depth`, `max_numeric_literal_digits`, `max_repeated_count`, + `max_big_float_exponent`; 0 for the default), and `marshal` refuses a value + nested past `max_nesting_depth`. Go's `encoding/pb` tests are ported with + their byte vectors, and the conformance tests check the bytes against the + protobuf runtime both ways. +- **`check_decode --format pb` decodes through `protowire.pb`**, into + dataclasses mirroring `adversarial.proto`, as Go's harness decodes through + its `encoding/pb`. It takes every `--limit` but `MaxBytesLiteralLength`. All + 29 of the corpus's `pb` rows pass, the 18 that skip Python today included, + with the same verdict as protowire-go's harness on each. + ### Changed - **`pxf.bind_row` and `DatasetReader.scan` bind a row as protowire-go does** diff --git a/README.md b/README.md index a119c9b..1b36628 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,42 @@ e = envelope.Err(400, "VALIDATION", "bad input").error.with_field( ) ``` +### PB for plain classes + +`protowire.pb` is a port of protowire-go's `encoding/pb`: protobuf binary for +dataclasses, with no `.proto` and no generated code. Each field names its +number, and its protobuf type where the Python type does not say it. The bytes +are protowire-go's, and decoding is bounded by the same per-call limits. + +```python +from dataclasses import dataclass, field +from decimal import Decimal +from typing import Annotated, Optional +from protowire import pb + +@dataclass +class Order: + id: Annotated[int, pb.Field(1, pb.UINT64)] = 0 + symbol: Annotated[str, pb.Field(2)] = "" + qty: Annotated[int, pb.Field(3, pb.SINT32)] = 0 # zigzag, Go's `protowire:"3,zigzag"` + price: Annotated[Optional[Decimal], pb.Field(4)] = None # pxf.Decimal + tags: Annotated[list[str], pb.Field(5)] = field(default_factory=list) + +data = pb.marshal(Order(id=7, symbol="ACME", qty=-100, price=Decimal("12.50"))) +order = pb.unmarshal(data, Order, max_message_size=1 << 20) + +# Length-prefixed streams, as Go's pb.Encoder / pb.Decoder frame them. +pb.Encoder(out).encode(order) +dec = pb.Decoder(inp) +while (o := dec.decode(Order)) is not None: + ... +``` + +`int` is int64 unless the field says otherwise (`pb.INT32`, `pb.UINT32`, +`pb.UINT64`, `pb.SINT32`, `pb.SINT64`, or `pb.BIGINT` for `pxf.BigInt`); +`float` is double unless `pb.FLOAT`. `decimal.Decimal` is `pxf.Decimal` and +`pb.BigFloat` is `pxf.BigFloat`. See the module docstring for the rest. + ## Build from source ```sh @@ -123,6 +159,11 @@ cross-port equivalence guarantees. - **No standalone Python CLI.** The shared CLI lives in [trendvidia/protowire/cmd/protowire](https://github.com/trendvidia/protowire/tree/main/cmd/protowire); Python callers either invoke that binary or use the in-process API. +- **`protowire.pb` is pure Python.** protowire-cpp's PB codec is C++ + templates over C++ structs, so there is nothing native to call on a + dataclass. It is slower than the native PXF and SBE paths. A message nested + past Python's recursion limit is a `ValueError`, so a `max_nesting_depth` + raised past a few hundred needs `sys.setrecursionlimit` raised with it. - **Free-threaded Python (PEP 703 / 3.13t)** is untested. nanobind supports it but the build hasn't been validated against `--disable-gil` interpreters. From 2f0cb8fa4480e8af4ff7b7d90a1890063f0d812e Mon Sep 17 00:00:00 2001 From: "Benedicto Franco Jr." Date: Sat, 7 Nov 2026 18:55:06 -0800 Subject: [PATCH 4/5] docs: 16 pb rows skip Python, not 18 --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 475287b..a1d9e51 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,7 +28,7 @@ format changes. - **`check_decode --format pb` decodes through `protowire.pb`**, into dataclasses mirroring `adversarial.proto`, as Go's harness decodes through its `encoding/pb`. It takes every `--limit` but `MaxBytesLiteralLength`. All - 29 of the corpus's `pb` rows pass, the 18 that skip Python today included, + 29 of the corpus's `pb` rows pass, the 16 that skip Python today included, with the same verdict as protowire-go's harness on each. ### Changed From 9d523b1812d8d91aa091fb9fe1a1e18b941bb419 Mon Sep 17 00:00:00 2001 From: "Benedicto Franco Jr." Date: Sat, 7 Nov 2026 18:57:53 -0800 Subject: [PATCH 5/5] pb: test NaN with math.isnan, and raise the decoder's stored error through a local CodeQL flagged v != v (comparison of identical values) and raising an Optional attribute (illegal raise of None). Same behaviour. --- src/protowire/pb.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/protowire/pb.py b/src/protowire/pb.py index f78a57c..15fb193 100644 --- a/src/protowire/pb.py +++ b/src/protowire/pb.py @@ -74,6 +74,7 @@ class Endpoint: import dataclasses import decimal +import math import struct import sys import types @@ -593,8 +594,8 @@ def _float_bits(k: Kind, v: Any) -> bytes: # A NaN is written as the canonical quiet NaN whatever its payload, as Go # writes it (draft -01 § Numeric Literals; protowire-java#161). if k is DOUBLE: - return b"\x00\x00\x00\x00\x00\x00\xf8\x7f" if v != v else struct.pack(" Exception: return e def decode(self, obj: Union[C, type[C]]) -> Optional[C]: - if self._err is not None: - raise self._err + err = self._err + if err is not None: + raise err # encoding/binary.ReadUvarint: ten bytes at most, the tenth one bit. n = 0 shift = 0