68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
"""Byte-wise cursor utilities for decoding binary ring payloads.
|
|
|
|
Every read is bounds-checked and raises :class:`ValueError` on a truncated or
|
|
malformed payload, so decoders surface a single, catchable error type for any
|
|
corruption (rather than leaking ``struct.error``/``UnicodeDecodeError``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import struct
|
|
|
|
|
|
class ByteCursor:
|
|
"""Read primitive values from bytes while tracking offset (bounds-checked)."""
|
|
|
|
def __init__(self, payload: bytes) -> None:
|
|
"""Create cursor at start of payload."""
|
|
self.payload = payload
|
|
self.offset = 0
|
|
|
|
def _take(self, size: int) -> bytes:
|
|
"""Consume ``size`` bytes, raising ValueError if the payload is too short."""
|
|
end = self.offset + size
|
|
if size < 0 or end > len(self.payload):
|
|
raise ValueError(
|
|
f"truncated payload: need {size} bytes at offset {self.offset}, "
|
|
f"only {len(self.payload) - self.offset} remain"
|
|
)
|
|
data = self.payload[self.offset : end]
|
|
self.offset = end
|
|
return data
|
|
|
|
def read_u8(self) -> int:
|
|
"""Read unsigned 8-bit integer."""
|
|
return struct.unpack("<B", self._take(1))[0]
|
|
|
|
def read_u16(self) -> int:
|
|
"""Read unsigned 16-bit integer."""
|
|
return struct.unpack("<H", self._take(2))[0]
|
|
|
|
def read_u32(self) -> int:
|
|
"""Read unsigned 32-bit integer."""
|
|
return struct.unpack("<I", self._take(4))[0]
|
|
|
|
def read_u64(self) -> int:
|
|
"""Read unsigned 64-bit integer."""
|
|
return struct.unpack("<Q", self._take(8))[0]
|
|
|
|
def read_f32(self) -> float:
|
|
"""Read 32-bit float."""
|
|
return float(struct.unpack("<f", self._take(4))[0])
|
|
|
|
def read_bytes(self, size: int) -> bytes:
|
|
"""Read raw byte slice of fixed size (bounds-checked)."""
|
|
return self._take(size)
|
|
|
|
def read_str(self, size: int) -> str:
|
|
"""Read a UTF-8 string of ``size`` bytes, raising ValueError on bad UTF-8."""
|
|
raw = self._take(size)
|
|
try:
|
|
return raw.decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
raise ValueError("invalid UTF-8 in payload string") from exc
|
|
|
|
def remaining_bytes(self) -> int:
|
|
"""Return unread byte count."""
|
|
return len(self.payload) - self.offset
|