Inserts a header just after the magic string of the provided flatbuffer data. Args: flatbuffer_data: The input data to modify. magic_regex: A regex pattern that must match the magic file_identifier characters of flatbuffer_data. header_data: The data to inser
(
flatbuffer_data: bytes, magic_regex: str, header_data: bytes
)
| 93 | |
| 94 | |
| 95 | def _insert_flatbuffer_header( |
| 96 | flatbuffer_data: bytes, magic_regex: str, header_data: bytes |
| 97 | ) -> bytes: |
| 98 | """Inserts a header just after the magic string of the provided flatbuffer data. |
| 99 | |
| 100 | Args: |
| 101 | flatbuffer_data: The input data to modify. |
| 102 | magic_regex: A regex pattern that must match the magic file_identifier |
| 103 | characters of flatbuffer_data. |
| 104 | header_data: The data to insert into flatbuffer_data. To ensure that |
| 105 | flatbuffer internal alignment is preserved, the caller must |
| 106 | guaranteed that its length is a power of 2 >= the largest |
| 107 | force_align value in the schema. |
| 108 | Returns: |
| 109 | The modified flatbuffer_data with header_data inserted. |
| 110 | Raises: |
| 111 | ValueError: If flatbuffer_data is too short to be valid. |
| 112 | ValueError: If the magic bytes of flatbuffer_data does not match |
| 113 | magic_regex. |
| 114 | """ |
| 115 | # The binary flatbuffer file should begin with: |
| 116 | # - Offset in bytes to root table (4 bytes little endian) |
| 117 | # - file_identifier string from the schema (4 bytes, string order) |
| 118 | if len(flatbuffer_data) < 8: |
| 119 | raise ValueError(f"Flatbuffer data length {len(flatbuffer_data)} < 8") |
| 120 | |
| 121 | # Ensure that the magic matches. |
| 122 | actual_magic: str = flatbuffer_data[4:8].decode(errors="replace") |
| 123 | if not re.match(magic_regex, actual_magic): |
| 124 | raise ValueError( |
| 125 | f"Flatbuffer data magic bytes {repr(actual_magic)} " |
| 126 | + f"does not match pattern /{magic_regex}/" |
| 127 | ) |
| 128 | |
| 129 | # Avoid a potentially big allocation/copy if there's nothing to do. |
| 130 | if len(header_data) == 0: |
| 131 | return flatbuffer_data |
| 132 | |
| 133 | # We will need to adjust the root object offset after inserting the header. |
| 134 | root_offset = int.from_bytes(flatbuffer_data[0:4], byteorder=_HEADER_BYTEORDER) |
| 135 | |
| 136 | return ( |
| 137 | # New root offset. |
| 138 | (root_offset + len(header_data)).to_bytes(4, byteorder=_HEADER_BYTEORDER) |
| 139 | # Existing magic bytes. |
| 140 | + flatbuffer_data[4:8] |
| 141 | # Provided header + padding. |
| 142 | + header_data |
| 143 | # Remainder of the file. Note that this can be O(10MB to 100MB), so it |
| 144 | # can trigger a large allocation + copy. |
| 145 | + flatbuffer_data[8:] |
| 146 | ) |
| 147 | |
| 148 | |
| 149 | @dataclass |
no test coverage detected