from __future__ import annotations import re import struct import sys from codecs import getincrementaldecoder from collections.abc import Callable, Generator, Mapping, Sequence from contextlib import contextmanager from datetime import date, datetime, timedelta, timezone from io import BytesIO from typing import IO, TYPE_CHECKING, Any, TypeVar, cast, overload from ._types import ( CBORDecodeEOF, CBORDecodeValueError, CBORSimpleValue, CBORTag, FrozenDict, break_marker, undefined, ) if TYPE_CHECKING: from decimal import Decimal from email.message import Message from fractions import Fraction from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network from typing import Literal from uuid import UUID T = TypeVar("T") timestamp_re = re.compile( r"^(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)" r"(?:\.(\d{1,6})\d*)?(?:Z|([+-])(\d\d):(\d\d))$" ) incremental_utf8_decoder = getincrementaldecoder("utf-8") class CBORDecoder: """ The CBORDecoder class implements a fully featured `CBOR`_ decoder with several extensions for handling shared references, big integers, rational numbers and so on. Typically the class is not used directly, but the :func:`load` and :func:`loads` functions are called to indirectly construct and use the class. When the class is constructed manually, the main entry points are :meth:`decode` and :meth:`decode_from_bytes`. .. _CBOR: https://cbor.io/ """ __slots__ = ( "_tag_hook", "_object_hook", "_share_index", "_shareables", "_fp", "_fp_read", "_immutable", "_str_errors", "_stringref_namespace", "_decode_depth", ) _fp: IO[bytes] _fp_read: Callable[[int], bytes] def __init__( self, fp: IO[bytes], tag_hook: Callable[[CBORDecoder, CBORTag], Any] | None = None, object_hook: Callable[[CBORDecoder, dict[Any, Any]], Any] | None = None, str_errors: Literal["strict", "error", "replace"] = "strict", ): """ :param fp: the file to read from (any file-like object opened for reading in binary mode) :param tag_hook: callable that takes 2 arguments: the decoder instance, and the :class:`.CBORTag` to be decoded. This callback is invoked for any tags for which there is no built-in decoder. The return value is substituted for the :class:`.CBORTag` object in the deserialized output :param object_hook: callable that takes 2 arguments: the decoder instance, and a dictionary. This callback is invoked for each deserialized :class:`dict` object. The return value is substituted for the dict in the deserialized output. :param str_errors: determines how to handle unicode decoding errors (see the `Error Handlers`_ section in the standard library documentation for details) .. _Error Handlers: https://docs.python.org/3/library/codecs.html#error-handlers """ self.fp = fp self.tag_hook = tag_hook self.object_hook = object_hook self.str_errors = str_errors self._share_index: int | None = None self._shareables: list[object] = [] self._stringref_namespace: list[str | bytes] | None = None self._immutable = False self._decode_depth = 0 @property def immutable(self) -> bool: """ Used by decoders to check if the calling context requires an immutable type. Object_hook or tag_hook should raise an exception if this flag is set unless the result can be safely used as a dict key. """ return self._immutable @property def fp(self) -> IO[bytes]: return self._fp @fp.setter def fp(self, value: IO[bytes]) -> None: try: if not callable(value.read): raise ValueError("fp.read is not callable") except AttributeError: raise ValueError("fp object has no read method") else: self._fp = value self._fp_read = value.read @property def tag_hook(self) -> Callable[[CBORDecoder, CBORTag], Any] | None: return self._tag_hook @tag_hook.setter def tag_hook(self, value: Callable[[CBORDecoder, CBORTag], Any] | None) -> None: if value is None or callable(value): self._tag_hook = value else: raise ValueError("tag_hook must be None or a callable") @property def object_hook(self) -> Callable[[CBORDecoder, dict[Any, Any]], Any] | None: return self._object_hook @object_hook.setter def object_hook(self, value: Callable[[CBORDecoder, dict[Any, Any]], Any] | None) -> None: if value is None or callable(value): self._object_hook = value else: raise ValueError("object_hook must be None or a callable") @property def str_errors(self) -> Literal["strict", "error", "replace"]: return self._str_errors @str_errors.setter def str_errors(self, value: Literal["strict", "error", "replace"]) -> None: if value in ("strict", "error", "replace"): self._str_errors = value else: raise ValueError( f"invalid str_errors value {value!r} (must be one of 'strict', " "'error', or 'replace')" ) def set_shareable(self, value: T) -> T: """ Set the shareable value for the last encountered shared value marker, if any. If the current shared index is ``None``, nothing is done. :param value: the shared value :returns: the shared value to permit chaining """ if self._share_index is not None: self._shareables[self._share_index] = value return value def _stringref_namespace_add(self, string: str | bytes, length: int) -> None: if self._stringref_namespace is not None: next_index = len(self._stringref_namespace) if next_index < 24: is_referenced = length >= 3 elif next_index < 256: is_referenced = length >= 4 elif next_index < 65536: is_referenced = length >= 5 elif next_index < 4294967296: is_referenced = length >= 7 else: is_referenced = length >= 11 if is_referenced: self._stringref_namespace.append(string) def read(self, amount: int) -> bytes: """ Read bytes from the data stream. :param int amount: the number of bytes to read """ data = self._fp_read(amount) if len(data) < amount: raise CBORDecodeEOF( f"premature end of stream (expected to read {amount} bytes, got {len(data)} " "instead)" ) return data def _decode(self, immutable: bool = False, unshared: bool = False) -> Any: if immutable: old_immutable = self._immutable self._immutable = True if unshared: old_index = self._share_index self._share_index = None try: initial_byte = self.read(1)[0] major_type = initial_byte >> 5 subtype = initial_byte & 31 decoder = major_decoders[major_type] return decoder(self, subtype) finally: if immutable: self._immutable = old_immutable if unshared: self._share_index = old_index @contextmanager def _decoding_context(self) -> Generator[None]: """ Context manager for tracking decode depth and clearing shared state. Shared state is cleared at the end of each top-level decode to prevent shared references from leaking between independent decode operations. Nested calls (from hooks) must preserve the state. """ self._decode_depth += 1 try: yield finally: self._decode_depth -= 1 assert self._decode_depth >= 0 if self._decode_depth == 0: self._shareables.clear() self._share_index = None def decode(self) -> object: """ Decode the next value from the stream. :raises CBORDecodeError: if there is any problem decoding the stream """ with self._decoding_context(): return self._decode() def decode_from_bytes(self, buf: bytes) -> object: """ Wrap the given bytestring as a file and call :meth:`decode` with it as the argument. This method was intended to be used from the ``tag_hook`` hook when an object needs to be decoded separately from the rest but while still taking advantage of the shared value registry. """ with self._decoding_context(): with BytesIO(buf) as fp: old_fp = self.fp self.fp = fp retval = self._decode() self.fp = old_fp return retval @overload def _decode_length(self, subtype: int) -> int: ... @overload def _decode_length(self, subtype: int, allow_indefinite: Literal[True]) -> int | None: ... def _decode_length(self, subtype: int, allow_indefinite: bool = False) -> int | None: if subtype < 24: return subtype elif subtype == 24: return self.read(1)[0] elif subtype == 25: return cast(int, struct.unpack(">H", self.read(2))[0]) elif subtype == 26: return cast(int, struct.unpack(">L", self.read(4))[0]) elif subtype == 27: return cast(int, struct.unpack(">Q", self.read(8))[0]) elif subtype == 31 and allow_indefinite: return None else: raise CBORDecodeValueError(f"unknown unsigned integer subtype 0x{subtype:x}") def decode_uint(self, subtype: int) -> int: # Major tag 0 return self.set_shareable(self._decode_length(subtype)) def decode_negint(self, subtype: int) -> int: # Major tag 1 return self.set_shareable(-self._decode_length(subtype) - 1) def decode_bytestring(self, subtype: int) -> bytes: # Major tag 2 length = self._decode_length(subtype, allow_indefinite=True) if length is None: # Indefinite length buf: list[bytes] = [] while True: initial_byte = self.read(1)[0] if initial_byte == 0xFF: result = b"".join(buf) break elif initial_byte >> 5 == 2: length = self._decode_length(initial_byte & 0x1F) if length is None or length > sys.maxsize: raise CBORDecodeValueError( f"invalid length for indefinite bytestring chunk 0x{length:x}" ) value = self.read(length) buf.append(value) else: raise CBORDecodeValueError( "non-bytestring found in indefinite length bytestring" ) else: if length > sys.maxsize: raise CBORDecodeValueError(f"invalid length for bytestring 0x{length:x}") elif length <= 65536: result = self.read(length) else: # Read large bytestrings 65536 (2 ** 16) bytes at a time left = length buffer = bytearray() while left: chunk_size = min(left, 65536) buffer.extend(self.read(chunk_size)) left -= chunk_size result = bytes(buffer) self._stringref_namespace_add(result, length) return self.set_shareable(result) def decode_string(self, subtype: int) -> str: # Major tag 3 length = self._decode_length(subtype, allow_indefinite=True) if length is None: # Indefinite length # NOTE: It may seem redundant to repeat this code to handle UTF-8 # strings but there is a reason to do this separately to # byte-strings. Specifically, the CBOR spec states (in sec. 2.2): # # Text strings with indefinite lengths act the same as byte # strings with indefinite lengths, except that all their chunks # MUST be definite-length text strings. Note that this implies # that the bytes of a single UTF-8 character cannot be spread # between chunks: a new chunk can only be started at a # character boundary. # # This precludes using the indefinite bytestring decoder above as # that would happily ignore UTF-8 characters split across chunks. buf: list[str] = [] while True: initial_byte = self.read(1)[0] if initial_byte == 0xFF: result = "".join(buf) break elif initial_byte >> 5 == 3: length = self._decode_length(initial_byte & 0x1F) if length is None or length > sys.maxsize: raise CBORDecodeValueError( f"invalid length for indefinite string chunk 0x{length:x}" ) try: value = self.read(length).decode("utf-8", self._str_errors) except UnicodeDecodeError as exc: raise CBORDecodeValueError("error decoding unicode string") from exc buf.append(value) else: raise CBORDecodeValueError("non-string found in indefinite length string") else: if length > sys.maxsize: raise CBORDecodeValueError(f"invalid length for string 0x{length:x}") if length <= 65536: try: result = self.read(length).decode("utf-8", self._str_errors) except UnicodeDecodeError as exc: raise CBORDecodeValueError("error decoding unicode string") from exc else: # Read and decode large text strings 65536 (2 ** 16) bytes at a time codec = incremental_utf8_decoder(self._str_errors) left = length result = "" while left: chunk_size = min(left, 65536) final = left <= chunk_size try: result += codec.decode(self.read(chunk_size), final) except UnicodeDecodeError as exc: raise CBORDecodeValueError("error decoding unicode string") from exc left -= chunk_size self._stringref_namespace_add(result, length) return self.set_shareable(result) def decode_array(self, subtype: int) -> Sequence[Any]: # Major tag 4 length = self._decode_length(subtype, allow_indefinite=True) if length is None: # Indefinite length items: list[Any] = [] if not self._immutable: self.set_shareable(items) while True: value = self._decode(unshared=True) if value is break_marker: break else: items.append(value) else: if length > sys.maxsize: raise CBORDecodeValueError(f"invalid length for array 0x{length:x}") items = [] if not self._immutable: self.set_shareable(items) for index in range(length): items.append(self._decode(unshared=True)) if self._immutable: items_tuple = tuple(items) self.set_shareable(items_tuple) return items_tuple return items def decode_map(self, subtype: int) -> Mapping[Any, Any]: # Major tag 5 length = self._decode_length(subtype, allow_indefinite=True) if length is None: # Indefinite length dictionary: dict[Any, Any] = {} self.set_shareable(dictionary) while True: key = self._decode(immutable=True, unshared=True) if key is break_marker: break else: dictionary[key] = self._decode(unshared=True) else: dictionary = {} self.set_shareable(dictionary) for _ in range(length): key = self._decode(immutable=True, unshared=True) dictionary[key] = self._decode(unshared=True) if self._object_hook: dictionary = self._object_hook(self, dictionary) self.set_shareable(dictionary) elif self._immutable: frozen_dict = FrozenDict(dictionary) self.set_shareable(dictionary) return frozen_dict return dictionary def decode_semantic(self, subtype: int) -> Any: # Major tag 6 tagnum = self._decode_length(subtype) if semantic_decoder := semantic_decoders.get(tagnum): return semantic_decoder(self) tag = CBORTag(tagnum, None) self.set_shareable(tag) tag.value = self._decode(unshared=True) if self._tag_hook: tag = self._tag_hook(self, tag) return self.set_shareable(tag) def decode_special(self, subtype: int) -> Any: # Simple value if subtype < 20: # XXX Set shareable? return CBORSimpleValue(subtype) # Major tag 7 try: return special_decoders[subtype](self) except KeyError as e: raise CBORDecodeValueError( f"Undefined Reserved major type 7 subtype 0x{subtype:x}" ) from e # # Semantic decoders (major tag 6) # def decode_epoch_date(self) -> date: # Semantic tag 100 value = self._decode() return self.set_shareable(date.fromordinal(value + 719163)) def decode_date_string(self) -> date: # Semantic tag 1004 value = self._decode() return self.set_shareable(date.fromisoformat(value)) def decode_datetime_string(self) -> datetime: # Semantic tag 0 value = self._decode() match = timestamp_re.match(value) if match: ( year, month, day, hour, minute, second, secfrac, offset_sign, offset_h, offset_m, ) = match.groups() if secfrac is None: microsecond = 0 else: microsecond = int(f"{secfrac:<06}") if offset_h: if offset_sign == "-": sign = -1 else: sign = 1 hours = int(offset_h) * sign minutes = int(offset_m) * sign tz = timezone(timedelta(hours=hours, minutes=minutes)) else: tz = timezone.utc return self.set_shareable( datetime( int(year), int(month), int(day), int(hour), int(minute), int(second), microsecond, tz, ) ) else: raise CBORDecodeValueError(f"invalid datetime string: {value!r}") def decode_epoch_datetime(self) -> datetime: # Semantic tag 1 value = self._decode() try: tmp = datetime.fromtimestamp(value, timezone.utc) except (OverflowError, OSError, ValueError) as exc: raise CBORDecodeValueError("error decoding datetime from epoch") from exc return self.set_shareable(tmp) def decode_positive_bignum(self) -> int: # Semantic tag 2 from binascii import hexlify value = self._decode() if not isinstance(value, bytes): raise CBORDecodeValueError("invalid bignum value " + str(value)) return self.set_shareable(int(hexlify(value), 16)) def decode_negative_bignum(self) -> int: # Semantic tag 3 return self.set_shareable(-self.decode_positive_bignum() - 1) def decode_fraction(self) -> Decimal: # Semantic tag 4 from decimal import Decimal try: exp, sig = self._decode() except (TypeError, ValueError) as e: raise CBORDecodeValueError("Incorrect tag 4 payload") from e tmp = Decimal(sig).as_tuple() return self.set_shareable(Decimal((tmp.sign, tmp.digits, exp))) def decode_bigfloat(self) -> Decimal: # Semantic tag 5 from decimal import Decimal try: exp, sig = self._decode() except (TypeError, ValueError) as e: raise CBORDecodeValueError("Incorrect tag 5 payload") from e return self.set_shareable(Decimal(sig) * (2 ** Decimal(exp))) def decode_stringref(self) -> str | bytes: # Semantic tag 25 if self._stringref_namespace is None: raise CBORDecodeValueError("string reference outside of namespace") index: int = self._decode() try: value = self._stringref_namespace[index] except IndexError: raise CBORDecodeValueError(f"string reference {index} not found") return value def decode_shareable(self) -> object: # Semantic tag 28 old_index = self._share_index self._share_index = len(self._shareables) self._shareables.append(None) try: return self._decode() finally: self._share_index = old_index def decode_sharedref(self) -> Any: # Semantic tag 29 value = self._decode(unshared=True) try: shared = self._shareables[value] except IndexError: raise CBORDecodeValueError(f"shared reference {value} not found") if shared is None: raise CBORDecodeValueError(f"shared value {value} has not been initialized") else: return shared def decode_complex(self) -> complex: # Semantic tag 43000 inputval = self._decode(immutable=True, unshared=True) try: value = complex(*inputval) except TypeError as exc: if not isinstance(inputval, tuple): raise CBORDecodeValueError( "error decoding complex: input value was not a tuple" ) from None raise CBORDecodeValueError("error decoding complex") from exc return self.set_shareable(value) def decode_rational(self) -> Fraction: # Semantic tag 30 from fractions import Fraction inputval = self._decode(immutable=True, unshared=True) try: value = Fraction(*inputval) except (TypeError, ZeroDivisionError) as exc: if not isinstance(inputval, tuple): raise CBORDecodeValueError( "error decoding rational: input value was not a tuple" ) from None raise CBORDecodeValueError("error decoding rational") from exc return self.set_shareable(value) def decode_regexp(self) -> re.Pattern[str]: # Semantic tag 35 try: value = re.compile(self._decode()) except re.error as exc: raise CBORDecodeValueError("error decoding regular expression") from exc return self.set_shareable(value) def decode_mime(self) -> Message: # Semantic tag 36 from email.parser import Parser try: value = Parser().parsestr(self._decode()) except TypeError as exc: raise CBORDecodeValueError("error decoding MIME message") from exc return self.set_shareable(value) def decode_uuid(self) -> UUID: # Semantic tag 37 from uuid import UUID try: value = UUID(bytes=self._decode()) except (TypeError, ValueError) as exc: raise CBORDecodeValueError("error decoding UUID value") from exc return self.set_shareable(value) def decode_stringref_namespace(self) -> Any: # Semantic tag 256 old_namespace = self._stringref_namespace self._stringref_namespace = [] value = self._decode() self._stringref_namespace = old_namespace return value def decode_set(self) -> set[Any] | frozenset[Any]: # Semantic tag 258 if self._immutable: return self.set_shareable(frozenset(self._decode(immutable=True))) else: return self.set_shareable(set(self._decode(immutable=True))) def decode_ipaddress(self) -> IPv4Address | IPv6Address | CBORTag: # Semantic tag 260 from ipaddress import ip_address buf = self.decode() if not isinstance(buf, bytes) or len(buf) not in (4, 6, 16): raise CBORDecodeValueError(f"invalid ipaddress value {buf!r}") elif len(buf) in (4, 16): return self.set_shareable(ip_address(buf)) elif len(buf) == 6: # MAC address return self.set_shareable(CBORTag(260, buf)) raise CBORDecodeValueError(f"invalid ipaddress value {buf!r}") def decode_ipnetwork(self) -> IPv4Network | IPv6Network: # Semantic tag 261 from ipaddress import ip_network net_map = self.decode() if isinstance(net_map, Mapping) and len(net_map) == 1: for net in net_map.items(): try: return self.set_shareable(ip_network(net, strict=False)) except (TypeError, ValueError): break raise CBORDecodeValueError(f"invalid ipnetwork value {net_map!r}") def decode_self_describe_cbor(self) -> Any: # Semantic tag 55799 return self._decode() # # Special decoders (major tag 7) # def decode_simple_value(self) -> CBORSimpleValue: # XXX Set shareable? return CBORSimpleValue(self.read(1)[0]) def decode_float16(self) -> float: return self.set_shareable(cast(float, struct.unpack(">e", self.read(2))[0])) def decode_float32(self) -> float: return self.set_shareable(cast(float, struct.unpack(">f", self.read(4))[0])) def decode_float64(self) -> float: return self.set_shareable(cast(float, struct.unpack(">d", self.read(8))[0])) major_decoders: dict[int, Callable[[CBORDecoder, int], Any]] = { 0: CBORDecoder.decode_uint, 1: CBORDecoder.decode_negint, 2: CBORDecoder.decode_bytestring, 3: CBORDecoder.decode_string, 4: CBORDecoder.decode_array, 5: CBORDecoder.decode_map, 6: CBORDecoder.decode_semantic, 7: CBORDecoder.decode_special, } special_decoders: dict[int, Callable[[CBORDecoder], Any]] = { 20: lambda self: False, 21: lambda self: True, 22: lambda self: None, 23: lambda self: undefined, 24: CBORDecoder.decode_simple_value, 25: CBORDecoder.decode_float16, 26: CBORDecoder.decode_float32, 27: CBORDecoder.decode_float64, 31: lambda self: break_marker, } semantic_decoders: dict[int, Callable[[CBORDecoder], Any]] = { 0: CBORDecoder.decode_datetime_string, 1: CBORDecoder.decode_epoch_datetime, 2: CBORDecoder.decode_positive_bignum, 3: CBORDecoder.decode_negative_bignum, 4: CBORDecoder.decode_fraction, 5: CBORDecoder.decode_bigfloat, 25: CBORDecoder.decode_stringref, 28: CBORDecoder.decode_shareable, 29: CBORDecoder.decode_sharedref, 30: CBORDecoder.decode_rational, 35: CBORDecoder.decode_regexp, 36: CBORDecoder.decode_mime, 37: CBORDecoder.decode_uuid, 100: CBORDecoder.decode_epoch_date, 256: CBORDecoder.decode_stringref_namespace, 258: CBORDecoder.decode_set, 260: CBORDecoder.decode_ipaddress, 261: CBORDecoder.decode_ipnetwork, 1004: CBORDecoder.decode_date_string, 43000: CBORDecoder.decode_complex, 55799: CBORDecoder.decode_self_describe_cbor, } def loads( s: bytes | bytearray | memoryview, tag_hook: Callable[[CBORDecoder, CBORTag], Any] | None = None, object_hook: Callable[[CBORDecoder, dict[Any, Any]], Any] | None = None, str_errors: Literal["strict", "error", "replace"] = "strict", ) -> Any: """ Deserialize an object from a bytestring. :param bytes s: the bytestring to deserialize :param tag_hook: callable that takes 2 arguments: the decoder instance, and the :class:`.CBORTag` to be decoded. This callback is invoked for any tags for which there is no built-in decoder. The return value is substituted for the :class:`.CBORTag` object in the deserialized output :param object_hook: callable that takes 2 arguments: the decoder instance, and a dictionary. This callback is invoked for each deserialized :class:`dict` object. The return value is substituted for the dict in the deserialized output. :param str_errors: determines how to handle unicode decoding errors (see the `Error Handlers`_ section in the standard library documentation for details) :return: the deserialized object .. _Error Handlers: https://docs.python.org/3/library/codecs.html#error-handlers """ with BytesIO(s) as fp: return CBORDecoder( fp, tag_hook=tag_hook, object_hook=object_hook, str_errors=str_errors ).decode() def load( fp: IO[bytes], tag_hook: Callable[[CBORDecoder, CBORTag], Any] | None = None, object_hook: Callable[[CBORDecoder, dict[Any, Any]], Any] | None = None, str_errors: Literal["strict", "error", "replace"] = "strict", ) -> Any: """ Deserialize an object from an open file. :param fp: the file to read from (any file-like object opened for reading in binary mode) :param tag_hook: callable that takes 2 arguments: the decoder instance, and the :class:`.CBORTag` to be decoded. This callback is invoked for any tags for which there is no built-in decoder. The return value is substituted for the :class:`.CBORTag` object in the deserialized output :param object_hook: callable that takes 2 arguments: the decoder instance, and a dictionary. This callback is invoked for each deserialized :class:`dict` object. The return value is substituted for the dict in the deserialized output. :param str_errors: determines how to handle unicode decoding errors (see the `Error Handlers`_ section in the standard library documentation for details) :return: the deserialized object .. _Error Handlers: https://docs.python.org/3/library/codecs.html#error-handlers """ return CBORDecoder( fp, tag_hook=tag_hook, object_hook=object_hook, str_errors=str_errors ).decode()