ef9ca9eeca
Final-review fixes for Plan 2 (fit-rewriter). Every failure mode below now surfaces as FitFormatError so Plan 3 can classify invalid FIT input as a non-retryable activity error (spec 10.4). - Range-check numeric values against the field's declared size before struct.pack, so an oversized serial number or a 1-byte product field raises FitFormatError instead of leaking a raw struct.error. - Reject zero-size field definitions during parsing. A zero-size device_info field 0 read back as device_index == 0 via int.from_bytes(b"", ...), which could have let a paired sensor be rewritten as an Edge 1030 Plus (spec 10.2). - Add DeviceFieldValue.is_creator so callers can tell the creator device_info record from sensor records instead of silently keeping whichever record appeared last. - Implement the missing spec 10.4 post-patch step: read the patched buffer back and verify file_id 1/2/8 and creator device_info 2/4/27 hold the target values. A field that could not be written (e.g. a product_name field too small for the target string) now fails the whole conversion rather than producing a silent partial patch. Verification runs before the output is written, so a half-rewritten file never lands on disk. - Use the field's actual endianness in _read_field_value's fallback path. - Add curated re-exports in app/fit/__init__.py for Plan 3. - Document _iter_data_fields' caller invariant (validate the container first; end_offset is not clamped). - Extend the preservation fixture with a product_name string field so the zero-filling string write path is covered by the byte-preservation proof, and test convert_fit_device against a 12-byte header. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
387 lines
15 KiB
Python
387 lines
15 KiB
Python
import struct
|
|
from pathlib import Path
|
|
|
|
from app.fit.crc import fit_crc
|
|
from app.fit.models import (
|
|
DeviceFieldValue,
|
|
FieldDefinition,
|
|
FitConversionResult,
|
|
GarminDevice,
|
|
LocalDefinition,
|
|
)
|
|
|
|
FILE_ID_MESG_NUM = 0
|
|
DEVICE_INFO_MESG_NUM = 23
|
|
|
|
# Device-identity fields read back / patched per message type.
|
|
FILE_ID_DEVICE_FIELDS = frozenset({1, 2, 3, 8})
|
|
DEVICE_INFO_DEVICE_FIELDS = frozenset({2, 3, 4, 27})
|
|
|
|
# Maximum unsigned value per declared FIT field size, used to reject values that
|
|
# cannot be represented in the field the source file actually declares.
|
|
_MAX_UNSIGNED_BY_SIZE = {1: 0xFF, 2: 0xFFFF, 4: 0xFFFFFFFF}
|
|
|
|
|
|
class FitFormatError(ValueError):
|
|
pass
|
|
|
|
|
|
def _validate_fit_container(data: bytearray) -> None:
|
|
if len(data) < 14:
|
|
raise FitFormatError("FIT file is too small")
|
|
header_size = data[0]
|
|
if header_size not in {12, 14}:
|
|
raise FitFormatError(f"Unsupported FIT header size: {header_size}")
|
|
if len(data) < header_size + 2:
|
|
raise FitFormatError("FIT file is shorter than its header")
|
|
if bytes(data[8:12]) != b".FIT":
|
|
raise FitFormatError("Missing .FIT signature")
|
|
data_size = struct.unpack_from("<I", data, 4)[0]
|
|
expected_size = header_size + data_size + 2
|
|
if len(data) != expected_size:
|
|
raise FitFormatError(f"FIT size mismatch: header says {expected_size} bytes, file has {len(data)}")
|
|
if header_size == 14:
|
|
expected_header_crc = struct.unpack_from("<H", data, 12)[0]
|
|
if expected_header_crc != fit_crc(data[:12]):
|
|
raise FitFormatError("FIT header CRC check failed")
|
|
expected_file_crc = struct.unpack_from("<H", data, len(data) - 2)[0]
|
|
if expected_file_crc != fit_crc(data[:-2]):
|
|
raise FitFormatError("FIT file CRC check failed")
|
|
|
|
|
|
def _read_definition(
|
|
data: bytearray,
|
|
offset: int,
|
|
has_developer_fields: bool,
|
|
end_offset: int,
|
|
) -> tuple[LocalDefinition, int]:
|
|
if offset + 5 > end_offset:
|
|
raise FitFormatError("Truncated FIT definition message")
|
|
offset += 1 # reserved byte
|
|
architecture = data[offset]
|
|
offset += 1
|
|
if architecture not in {0, 1}:
|
|
raise FitFormatError(f"Unsupported FIT architecture: {architecture}")
|
|
endian = ">" if architecture == 1 else "<"
|
|
global_message_num = struct.unpack_from(f"{endian}H", data, offset)[0]
|
|
offset += 2
|
|
field_count = data[offset]
|
|
offset += 1
|
|
|
|
fields: list[FieldDefinition] = []
|
|
for _ in range(field_count):
|
|
if offset + 3 > end_offset:
|
|
raise FitFormatError("Truncated FIT field definition")
|
|
field_num = data[offset]
|
|
field_size = data[offset + 1]
|
|
# A zero-size field is illegal FIT. Rejecting it here (rather than
|
|
# special-casing it downstream) closes a hole where a zero-size
|
|
# device_info field 0 would read back as device_index == 0 via
|
|
# int.from_bytes(b"", ...) and make a sensor record look like the creator.
|
|
if field_size == 0:
|
|
raise FitFormatError(f"FIT field {field_num} declares an invalid size of 0")
|
|
fields.append(FieldDefinition(field_num, field_size, data[offset + 2]))
|
|
offset += 3
|
|
|
|
developer_field_size = 0
|
|
if has_developer_fields:
|
|
if offset >= end_offset:
|
|
raise FitFormatError("Truncated FIT developer field count")
|
|
developer_count = data[offset]
|
|
offset += 1
|
|
for _ in range(developer_count):
|
|
if offset + 3 > end_offset:
|
|
raise FitFormatError("Truncated FIT developer field definition")
|
|
developer_field_size += data[offset + 1]
|
|
offset += 3
|
|
|
|
return LocalDefinition(global_message_num, endian, tuple(fields), developer_field_size), offset
|
|
|
|
|
|
def _collect_field_offsets(
|
|
definition: LocalDefinition,
|
|
offset: int,
|
|
end_offset: int,
|
|
) -> tuple[list[tuple[FieldDefinition, int]], int]:
|
|
result: list[tuple[FieldDefinition, int]] = []
|
|
current = offset
|
|
for field in definition.fields:
|
|
if current + field.size > end_offset:
|
|
raise FitFormatError("Truncated FIT data record")
|
|
result.append((field, current))
|
|
current += field.size
|
|
if current + definition.developer_field_size > end_offset:
|
|
raise FitFormatError("Truncated FIT developer field payload")
|
|
current += definition.developer_field_size
|
|
return result, current
|
|
|
|
|
|
def _iter_data_fields(
|
|
data: bytearray,
|
|
) -> list[tuple[LocalDefinition, list[tuple[FieldDefinition, int]]]]:
|
|
"""Walk the FIT data section and return every data record with its field offsets.
|
|
|
|
Caller invariant: callers MUST run ``_validate_fit_container(data)`` first. The
|
|
header-derived ``end_offset`` is trusted as-is and never clamped to ``len(data)``,
|
|
so an unvalidated buffer whose declared data size exceeds its real length would be
|
|
parsed out of bounds instead of rejected cleanly.
|
|
"""
|
|
header_size = data[0]
|
|
data_size = struct.unpack_from("<I", data, 4)[0]
|
|
offset = header_size
|
|
end_offset = header_size + data_size
|
|
definitions: dict[int, LocalDefinition] = {}
|
|
records: list[tuple[LocalDefinition, list[tuple[FieldDefinition, int]]]] = []
|
|
|
|
while offset < end_offset:
|
|
record_header = data[offset]
|
|
offset += 1
|
|
|
|
if record_header & 0x80:
|
|
local = (record_header >> 5) & 0x03
|
|
definition = definitions.get(local)
|
|
if definition is None:
|
|
raise FitFormatError(f"Compressed timestamp record used unknown local definition {local}")
|
|
field_offsets, offset = _collect_field_offsets(definition, offset, end_offset)
|
|
records.append((definition, field_offsets))
|
|
continue
|
|
|
|
local = record_header & 0x0F
|
|
is_definition = bool(record_header & 0x40)
|
|
has_developer_fields = bool(record_header & 0x20)
|
|
if is_definition:
|
|
definition, offset = _read_definition(data, offset, has_developer_fields, end_offset)
|
|
definitions[local] = definition
|
|
continue
|
|
|
|
definition = definitions.get(local)
|
|
if definition is None:
|
|
raise FitFormatError(f"Data record used unknown local definition {local}")
|
|
field_offsets, offset = _collect_field_offsets(definition, offset, end_offset)
|
|
records.append((definition, field_offsets))
|
|
|
|
if offset != end_offset:
|
|
raise FitFormatError("FIT parser did not end on data boundary")
|
|
return records
|
|
|
|
|
|
def is_fit_file(path: Path) -> bool:
|
|
try:
|
|
data = bytearray(path.read_bytes())
|
|
_validate_fit_container(data)
|
|
_iter_data_fields(data)
|
|
except (OSError, FitFormatError):
|
|
return False
|
|
return True
|
|
|
|
|
|
def convert_fit_device(
|
|
source_path: Path, output_path: Path, device: GarminDevice | None = None
|
|
) -> FitConversionResult:
|
|
resolved = device or GarminDevice()
|
|
data = bytearray(source_path.read_bytes())
|
|
_validate_fit_container(data)
|
|
patched_count = _patch_device_metadata(data, resolved)
|
|
header_crc = _rewrite_header_crc(data)
|
|
file_crc = _rewrite_file_crc(data)
|
|
_validate_fit_container(data)
|
|
_verify_patched_metadata(data, resolved)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_bytes(data)
|
|
return FitConversionResult(source_path, output_path, patched_count, header_crc, file_crc)
|
|
|
|
|
|
def read_device_field_values(path: Path) -> list[DeviceFieldValue]:
|
|
data = bytearray(path.read_bytes())
|
|
_validate_fit_container(data)
|
|
return _read_device_field_values(data)
|
|
|
|
|
|
def _read_device_field_values(data: bytearray) -> list[DeviceFieldValue]:
|
|
values: list[DeviceFieldValue] = []
|
|
for definition, field_offsets in _iter_data_fields(data):
|
|
if definition.global_message_num == FILE_ID_MESG_NUM:
|
|
interesting_fields = FILE_ID_DEVICE_FIELDS
|
|
is_creator = True
|
|
elif definition.global_message_num == DEVICE_INFO_MESG_NUM:
|
|
interesting_fields = DEVICE_INFO_DEVICE_FIELDS
|
|
field_map = {field.num: (field, offset) for field, offset in field_offsets}
|
|
is_creator = _is_creator_device_info(data, definition, field_map)
|
|
else:
|
|
continue
|
|
|
|
for field, offset in field_offsets:
|
|
if field.num not in interesting_fields:
|
|
continue
|
|
values.append(
|
|
DeviceFieldValue(
|
|
definition.global_message_num,
|
|
field.num,
|
|
_read_field_value(data, offset, field, definition.endian),
|
|
is_creator,
|
|
)
|
|
)
|
|
return values
|
|
|
|
|
|
def _is_creator_device_info(
|
|
data: bytearray,
|
|
definition: LocalDefinition,
|
|
field_map: dict[int, tuple[FieldDefinition, int]],
|
|
) -> bool:
|
|
"""A ``device_info`` record counts as the creator only when it carries an
|
|
explicit device_index (field 0) equal to 0. Records without field 0 are never
|
|
treated as the creator, so paired sensors are never rewritten."""
|
|
device_index_entry = field_map.get(0)
|
|
if device_index_entry is None:
|
|
return False
|
|
index_field, index_offset = device_index_entry
|
|
return _read_field_value(data, index_offset, index_field, definition.endian) == 0
|
|
|
|
|
|
def _patch_device_metadata(data: bytearray, device: GarminDevice) -> int:
|
|
patched_count = 0
|
|
eligible_field_count = 0
|
|
for definition, field_offsets in _iter_data_fields(data):
|
|
field_map = {field.num: (field, offset) for field, offset in field_offsets}
|
|
if definition.global_message_num == DEVICE_INFO_MESG_NUM and not _is_creator_device_info(
|
|
data, definition, field_map
|
|
):
|
|
continue
|
|
|
|
for field, offset in field_offsets:
|
|
target_value: int | str | None = None
|
|
if definition.global_message_num == FILE_ID_MESG_NUM:
|
|
if field.num == 1:
|
|
target_value = device.manufacturer_id
|
|
elif field.num == 2:
|
|
target_value = device.product_id
|
|
elif field.num == 3 and device.serial_number is not None:
|
|
target_value = device.serial_number
|
|
elif field.num == 8:
|
|
target_value = device.product_name
|
|
elif definition.global_message_num == DEVICE_INFO_MESG_NUM:
|
|
if field.num == 2:
|
|
target_value = device.manufacturer_id
|
|
elif field.num == 3 and device.serial_number is not None:
|
|
target_value = device.serial_number
|
|
elif field.num == 4:
|
|
target_value = device.product_id
|
|
elif field.num == 27:
|
|
target_value = device.product_name
|
|
|
|
if target_value is not None:
|
|
eligible_field_count += 1
|
|
if _write_field_value(data, offset, field, definition.endian, target_value):
|
|
patched_count += 1
|
|
|
|
if eligible_field_count == 0:
|
|
raise FitFormatError("No writable file_id or device_info device fields found")
|
|
return patched_count
|
|
|
|
|
|
def _expected_target_value(
|
|
global_message_num: int, field_num: int, device: GarminDevice
|
|
) -> int | str | None:
|
|
"""Target value a patched device-identity field must read back as, or None for
|
|
fields that are not verified (e.g. the optional serial number)."""
|
|
if global_message_num == FILE_ID_MESG_NUM:
|
|
return {1: device.manufacturer_id, 2: device.product_id, 8: device.product_name}.get(field_num)
|
|
if global_message_num == DEVICE_INFO_MESG_NUM:
|
|
return {2: device.manufacturer_id, 4: device.product_id, 27: device.product_name}.get(field_num)
|
|
return None
|
|
|
|
|
|
def _verify_patched_metadata(data: bytearray, device: GarminDevice) -> None:
|
|
"""Spec 10.4 post-patch step: verify the expected target metadata is readable.
|
|
|
|
Reads the patched buffer back and confirms every device-identity field that was
|
|
present in the source (file_id 1/2/8 and creator device_info 2/4/27) now holds
|
|
the target value. A field that could not be written -- e.g. a product_name field
|
|
too small for the target string -- fails the whole conversion with FitFormatError
|
|
instead of silently producing a half-rewritten file.
|
|
"""
|
|
for value in _read_device_field_values(data):
|
|
if not value.is_creator:
|
|
continue
|
|
expected = _expected_target_value(value.global_message_num, value.field_num, device)
|
|
if expected is None:
|
|
continue
|
|
if value.value != expected:
|
|
raise FitFormatError(
|
|
f"Patched FIT metadata verification failed for message "
|
|
f"{value.global_message_num} field {value.field_num}: "
|
|
f"expected {expected!r}, read back {value.value!r}"
|
|
)
|
|
|
|
|
|
def _read_field_value(data: bytearray, offset: int, field: FieldDefinition, endian: str) -> int | str:
|
|
base_type = field.base_type & 0x1F
|
|
if base_type in {0x03, 0x04, 0x0B} and field.size >= 2:
|
|
return struct.unpack_from(f"{endian}H", data, offset)[0]
|
|
if base_type in {0x05, 0x06, 0x0C} and field.size >= 4:
|
|
return struct.unpack_from(f"{endian}I", data, offset)[0]
|
|
if base_type == 0x07:
|
|
raw = bytes(data[offset : offset + field.size])
|
|
if 0 in raw:
|
|
raw = raw[: raw.index(0)]
|
|
return raw.decode("utf-8", errors="replace")
|
|
raw = bytes(data[offset : offset + field.size])
|
|
return int.from_bytes(raw, "little" if endian == "<" else "big")
|
|
|
|
|
|
def _write_field_value(
|
|
data: bytearray,
|
|
offset: int,
|
|
field: FieldDefinition,
|
|
endian: str,
|
|
value: int | str,
|
|
) -> bool:
|
|
if isinstance(value, str):
|
|
encoded = value.encode("utf-8")
|
|
if not encoded or field.size == 0 or len(encoded) + 1 > field.size:
|
|
return False
|
|
replacement = encoded + b"\x00" + b"\x00" * (field.size - len(encoded) - 1)
|
|
if bytes(data[offset : offset + field.size]) == replacement:
|
|
return False
|
|
data[offset : offset + field.size] = replacement
|
|
return True
|
|
|
|
max_value = _MAX_UNSIGNED_BY_SIZE.get(field.size)
|
|
if max_value is None:
|
|
return False
|
|
# Range-check before packing: struct.pack would otherwise raise a raw
|
|
# struct.error, which callers cannot classify alongside FitFormatError.
|
|
if not isinstance(value, int) or value < 0 or value > max_value:
|
|
raise FitFormatError(
|
|
f"Value {value!r} does not fit FIT field {field.num} "
|
|
f"of declared size {field.size} (allowed range 0-{max_value})"
|
|
)
|
|
|
|
if field.size == 1:
|
|
replacement = struct.pack("B", value)
|
|
elif field.size == 2:
|
|
replacement = struct.pack(f"{endian}H", value)
|
|
else:
|
|
replacement = struct.pack(f"{endian}I", value)
|
|
|
|
if bytes(data[offset : offset + field.size]) == replacement:
|
|
return False
|
|
data[offset : offset + field.size] = replacement
|
|
return True
|
|
|
|
|
|
def _rewrite_header_crc(data: bytearray) -> int | None:
|
|
header_size = data[0]
|
|
if header_size != 14:
|
|
return None
|
|
header_crc = fit_crc(data[:12])
|
|
struct.pack_into("<H", data, 12, header_crc)
|
|
return header_crc
|
|
|
|
|
|
def _rewrite_file_crc(data: bytearray) -> int:
|
|
file_crc = fit_crc(data[:-2])
|
|
struct.pack_into("<H", data, len(data) - 2, file_crc)
|
|
return file_crc
|