| # Copyright 2026 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import struct |
| import typing |
| |
| |
| def _align_to_8(offset: int) -> int: |
| """Rounds an offset up to the next 8-byte boundary. |
| |
| According to the FIDL wire format specification (RFC-0027/RFC-0120), all |
| out-of-line secondary objects (such as strings, vectors, out-of-line envelopes, |
| and out-of-line structs) must be aligned to 8-byte boundaries. |
| """ |
| pad_mod = offset % 8 |
| return offset + (8 - pad_mod) if pad_mod != 0 else offset |
| |
| |
| class Decoder: |
| def __init__(self, buf: bytes, handles: list[typing.Any]) -> None: |
| self.buf = buf |
| self.handles = handles |
| self.ool_offset = 0 |
| self.handle_index = 0 |
| |
| def _align_ool_offset(self) -> None: |
| """Aligns the out-of-line offset to the next 8-byte boundary.""" |
| self.ool_offset = _align_to_8(self.ool_offset) |
| |
| def read_handle(self, offset: int, nullable: bool = False) -> typing.Any: |
| presence = self._read_primitive("<I", offset) |
| if presence == 0: |
| if nullable: |
| return None |
| else: |
| raise ValueError("Absent non-nullable handle") |
| if self.handle_index >= len(self.handles): |
| raise ValueError("Not enough handles in decoder") |
| hdl = self.handles[self.handle_index] |
| self.handle_index += 1 |
| # Handles in `self.handles` can either be handle wrappers (e.g., |
| # `fuchsia_controller_py.Handle` or FDomain handles providing `.as_int()`) |
| # or raw integer handle values (e.g., in unit tests or GIDL runners). |
| if hasattr(hdl, "as_int"): |
| return hdl.as_int() |
| return int(hdl) |
| |
| def read_bool(self, offset: int) -> bool: |
| return self.buf[offset] != 0 |
| |
| def _read_primitive(self, fmt: str, offset: int) -> typing.Any: |
| return struct.unpack_from(fmt, self.buf, offset)[0] |
| |
| def read_int8(self, offset: int) -> int: |
| return self._read_primitive("<b", offset) |
| |
| def read_uint8(self, offset: int) -> int: |
| return self._read_primitive("<B", offset) |
| |
| def read_int16(self, offset: int) -> int: |
| return self._read_primitive("<h", offset) |
| |
| def read_uint16(self, offset: int) -> int: |
| return self._read_primitive("<H", offset) |
| |
| def read_int32(self, offset: int) -> int: |
| return self._read_primitive("<i", offset) |
| |
| def read_uint32(self, offset: int) -> int: |
| return self._read_primitive("<I", offset) |
| |
| def read_int64(self, offset: int) -> int: |
| return self._read_primitive("<q", offset) |
| |
| def read_uint64(self, offset: int) -> int: |
| return self._read_primitive("<Q", offset) |
| |
| def read_float32(self, offset: int) -> float: |
| return self._read_primitive("<f", offset) |
| |
| def read_float64(self, offset: int) -> float: |
| return self._read_primitive("<d", offset) |
| |
| def read_string(self, offset: int, nullable: bool = False) -> typing.Any: |
| length = struct.unpack_from("<Q", self.buf, offset)[0] |
| presence = struct.unpack_from("<Q", self.buf, offset + 8)[0] |
| if presence == 0: |
| if nullable: |
| return None |
| else: |
| raise ValueError("Absent non-nullable string") |
| self._align_ool_offset() |
| res = self.buf[self.ool_offset : self.ool_offset + length].decode( |
| "utf-8" |
| ) |
| self.ool_offset += length |
| return res |
| |
| def read_vector( |
| self, |
| offset: int, |
| element_decoder_fn: typing.Callable[["Decoder", int], typing.Any], |
| element_size: int, |
| nullable: bool = False, |
| ) -> typing.Any: |
| count = struct.unpack_from("<Q", self.buf, offset)[0] |
| presence = struct.unpack_from("<Q", self.buf, offset + 8)[0] |
| if presence == 0: |
| if nullable: |
| return None |
| else: |
| raise ValueError("Absent non-nullable vector") |
| self._align_ool_offset() |
| start_offset = self.ool_offset |
| self.ool_offset += element_size * count |
| res = [] |
| for i in range(count): |
| res.append( |
| element_decoder_fn(self, start_offset + i * element_size) |
| ) |
| return res |
| |
| def read_array( |
| self, |
| offset: int, |
| element_decoder_fn: typing.Callable[["Decoder", int], typing.Any], |
| element_size: int, |
| count: int, |
| ) -> list[typing.Any]: |
| res = [] |
| for i in range(count): |
| res.append(element_decoder_fn(self, offset + i * element_size)) |
| return res |
| |
| def read_optional_struct( |
| self, |
| offset: int, |
| decode_fn: typing.Callable[["Decoder", int], typing.Any], |
| inline_size: int, |
| ) -> typing.Any: |
| presence = struct.unpack_from("<Q", self.buf, offset)[0] |
| if presence == 0: |
| return None |
| |
| self._align_ool_offset() |
| start_offset = self.ool_offset |
| self.ool_offset += inline_size |
| return decode_fn(self, start_offset) |
| |
| def read_envelope( |
| self, |
| env_offset: int, |
| decode_fn: typing.Callable[["Decoder", int], typing.Any], |
| inline_size: int, |
| ) -> typing.Any: |
| is_empty = self.buf[env_offset : env_offset + 8] == b"\x00" * 8 |
| if is_empty: |
| return None |
| |
| size_or_value = self.buf[env_offset : env_offset + 4] |
| struct.unpack_from("<H", self.buf, env_offset + 4)[0] |
| flags = struct.unpack_from("<H", self.buf, env_offset + 6)[0] |
| inlined = (flags & 1) != 0 |
| |
| if inlined: |
| temp_decoder = Decoder( |
| size_or_value, self.handles[self.handle_index :] |
| ) |
| val = decode_fn(temp_decoder, 0) |
| self.handle_index += temp_decoder.handle_index |
| return val |
| else: |
| self._align_ool_offset() |
| num_bytes = struct.unpack_from("<I", self.buf, env_offset)[0] |
| member_offset = self.ool_offset |
| |
| inline_size_aligned = _align_to_8(inline_size) |
| |
| self.ool_offset = member_offset + inline_size_aligned |
| val = decode_fn(self, member_offset) |
| |
| self.ool_offset = _align_to_8(member_offset + num_bytes) |
| return val |
| |
| def skip_envelope(self, env_offset: int) -> None: |
| is_empty = self.buf[env_offset : env_offset + 8] == b"\x00" * 8 |
| if is_empty: |
| return |
| |
| num_handles = struct.unpack_from("<H", self.buf, env_offset + 4)[0] |
| flags = struct.unpack_from("<H", self.buf, env_offset + 6)[0] |
| inlined = (flags & 1) != 0 |
| |
| self.handle_index += num_handles |
| |
| if not inlined: |
| num_bytes = struct.unpack_from("<I", self.buf, env_offset)[0] |
| self._align_ool_offset() |
| self.ool_offset = _align_to_8(self.ool_offset + num_bytes) |
| |
| |
| def decode_struct( |
| bytes: bytes, |
| handles: list[typing.Any], |
| decode_fn: typing.Callable[[Decoder, int], typing.Any], |
| inline_size: int, |
| ) -> typing.Any: |
| decoder = Decoder(bytes, handles) |
| decoder.ool_offset = _align_to_8(inline_size) |
| return decode_fn(decoder, 0) |
