diff --git a/app/fit/__init__.py b/app/fit/__init__.py index e69de29..4b7389d 100644 --- a/app/fit/__init__.py +++ b/app/fit/__init__.py @@ -0,0 +1,17 @@ +from app.fit.models import DeviceFieldValue, FitConversionResult, GarminDevice +from app.fit.rewriter import ( + FitFormatError, + convert_fit_device, + is_fit_file, + read_device_field_values, +) + +__all__ = [ + "DeviceFieldValue", + "FitConversionResult", + "FitFormatError", + "GarminDevice", + "convert_fit_device", + "is_fit_file", + "read_device_field_values", +] diff --git a/app/fit/models.py b/app/fit/models.py index 4b85484..11f9771 100644 --- a/app/fit/models.py +++ b/app/fit/models.py @@ -42,3 +42,7 @@ class DeviceFieldValue: global_message_num: int field_num: int value: int | str + #: True when this value came from the creator device record. ``file_id`` + #: values are always creator values (a FIT file has exactly one file_id); + #: ``device_info`` values are creator values only when device_index == 0. + is_creator: bool diff --git a/app/fit/rewriter.py b/app/fit/rewriter.py index 2680c8d..ae4ea0e 100644 --- a/app/fit/rewriter.py +++ b/app/fit/rewriter.py @@ -13,6 +13,14 @@ from app.fit.models import ( 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 @@ -64,7 +72,15 @@ def _read_definition( for _ in range(field_count): if offset + 3 > end_offset: raise FitFormatError("Truncated FIT field definition") - fields.append(FieldDefinition(data[offset], data[offset + 1], data[offset + 2])) + 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 @@ -103,6 +119,13 @@ def _collect_field_offsets( 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(" 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 definition.global_message_num == FILE_ID_MESG_NUM and field.num in {1, 2, 3, 8}: - values.append( - DeviceFieldValue( - definition.global_message_num, - field.num, - _read_field_value(data, offset, field, definition.endian), - ) - ) - if definition.global_message_num == DEVICE_INFO_MESG_NUM and field.num in { - 2, - 3, - 4, - 27, - }: - values.append( - DeviceFieldValue( - definition.global_message_num, - field.num, - _read_field_value(data, offset, field, definition.endian), - ) + 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: - device_index_entry = field_map.get(0) - if device_index_entry is None: - continue - index_field, index_offset = device_index_entry - device_index = _read_field_value(data, index_offset, index_field, definition.endian) - if device_index != 0: - continue + 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 @@ -242,6 +280,41 @@ def _patch_device_metadata(data: bytearray, device: GarminDevice) -> int: 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: @@ -254,7 +327,7 @@ def _read_field_value(data: bytearray, offset: int, field: FieldDefinition, endi 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") + return int.from_bytes(raw, "little" if endian == "<" else "big") def _write_field_value( @@ -274,14 +347,23 @@ def _write_field_value( 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) - elif field.size == 4: - replacement = struct.pack(f"{endian}I", value) else: - return False + replacement = struct.pack(f"{endian}I", value) if bytes(data[offset : offset + field.size]) == replacement: return False diff --git a/tests/fit/test_rewriter_patching.py b/tests/fit/test_rewriter_patching.py index 0dccbfb..550a617 100644 --- a/tests/fit/test_rewriter_patching.py +++ b/tests/fit/test_rewriter_patching.py @@ -8,6 +8,7 @@ from app.fit.rewriter import ( FitFormatError, _iter_data_fields, convert_fit_device, + is_fit_file, read_device_field_values, ) from tests.fit.builders import data, definition, make_fit @@ -74,6 +75,44 @@ def _build_product_name_fixture() -> bytes: return make_fit(records) +def _build_zero_size_device_index_fixture() -> bytes: + """Adversarial file declaring device_info field 0 (device_index) with size 0. + + Without an explicit rejection, `int.from_bytes(b"", "little") == 0` would make + every record on this definition look like the creator (device_index == 0) and a + paired sensor would be rewritten as an Edge 1030 Plus.""" + file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)]) + file_data = data(0, struct.pack(" bytes: + """file_id with a 2-byte serial_number field (3/u16) -- too small to hold a + 32-bit serial number handed in through the public API.""" + file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84), (3, 2, 0x84)]) + file_data = data(0, struct.pack(" bytes: + """file_id declaring product (field 2) as a single byte, which cannot hold 3570.""" + file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 1, 0x02)]) + file_data = data(0, struct.pack(" bytes: + """file_id product_name field of size 10 -- too small for "Edge 1030 Plus" + (14 bytes plus a null terminator), so the string write silently no-ops.""" + file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84), (8, 10, 0x07)]) + file_data = data(0, struct.pack(" list[dict[int, int]]: """Parse raw bytes and return one dict of {field_num: value} per device_info record, in file order, so creator and sensor records can be distinguished positionally.""" @@ -214,3 +253,93 @@ def test_read_device_field_values_returns_device_field_value_instances(tmp_path: assert all(isinstance(v, DeviceFieldValue) for v in values) assert any(v.global_message_num == FILE_ID_MESG_NUM for v in values) + + +def test_zero_size_field_definition_is_rejected(tmp_path: Path) -> None: + """A zero-size field is illegal FIT and must be rejected during parsing, so a + zero-size device_index can never make a sensor record read as the creator.""" + source = tmp_path / "zero-size.fit" + source.write_bytes(_build_zero_size_device_index_fixture()) + output = tmp_path / "output.fit" + + assert is_fit_file(source) is False + with pytest.raises(FitFormatError): + convert_fit_device(source, output) + assert not output.exists() + + +def test_value_too_large_for_declared_field_size_raises_fit_format_error(tmp_path: Path) -> None: + """A 32-bit serial number against a 2-byte serial field must surface as + FitFormatError, not a raw struct.error escaping the public API.""" + source = tmp_path / "source.fit" + source.write_bytes(_build_undersized_serial_fixture()) + output = tmp_path / "output.fit" + + with pytest.raises(FitFormatError): + convert_fit_device(source, output, GarminDevice(serial_number=4294967295)) + assert not output.exists() + + +def test_product_id_too_large_for_one_byte_field_raises_fit_format_error(tmp_path: Path) -> None: + source = tmp_path / "source.fit" + source.write_bytes(_build_undersized_product_fixture()) + output = tmp_path / "output.fit" + + with pytest.raises(FitFormatError): + convert_fit_device(source, output) + assert not output.exists() + + +def test_unwritable_product_name_fails_conversion(tmp_path: Path) -> None: + """A product_name field too small for the target string used to be silently left + unpatched while manufacturer/product reported success. The post-patch read-back + verification must turn that partial patch into a controlled failure.""" + source = tmp_path / "source.fit" + source.write_bytes(_build_undersized_product_name_fixture()) + output = tmp_path / "output.fit" + + with pytest.raises(FitFormatError): + convert_fit_device(source, output) + assert not output.exists() + + +def test_read_device_field_values_marks_creator_records(tmp_path: Path) -> None: + """The creator device_info and every file_id field are is_creator=True; a paired + sensor's device_info fields are is_creator=False, so a caller building a + {(mesg, field): value} dict can no longer be shadowed by sensor values.""" + source = tmp_path / "source.fit" + source.write_bytes(_build_fixture()) + output = tmp_path / "output.fit" + + convert_fit_device(source, output) + + values = read_device_field_values(output) + + assert all(v.is_creator for v in values if v.global_message_num == FILE_ID_MESG_NUM) + + creator = { + (v.global_message_num, v.field_num): v.value + for v in values + if v.global_message_num == DEVICE_INFO_MESG_NUM and v.is_creator + } + sensor = { + (v.global_message_num, v.field_num): v.value + for v in values + if v.global_message_num == DEVICE_INFO_MESG_NUM and not v.is_creator + } + + assert creator[(DEVICE_INFO_MESG_NUM, 2)] == 1 + assert creator[(DEVICE_INFO_MESG_NUM, 4)] == 3570 + assert sensor[(DEVICE_INFO_MESG_NUM, 2)] == 32 + assert sensor[(DEVICE_INFO_MESG_NUM, 4)] == 1234 + + +def test_device_info_without_device_index_is_not_creator(tmp_path: Path) -> None: + source = tmp_path / "source.fit" + source.write_bytes(_build_no_device_index_fixture()) + + values = read_device_field_values(source) + + device_info_values = [v for v in values if v.global_message_num == DEVICE_INFO_MESG_NUM] + assert device_info_values + assert all(v.is_creator is False for v in device_info_values) diff --git a/tests/fit/test_rewriter_preservation.py b/tests/fit/test_rewriter_preservation.py index 35ff1b3..8c5d1d2 100644 --- a/tests/fit/test_rewriter_preservation.py +++ b/tests/fit/test_rewriter_preservation.py @@ -11,6 +11,7 @@ FILE_ID_MESG_NUM = 0 DEVICE_INFO_MESG_NUM = 23 RECORD_MESG_NUM = 20 HEADER_SIZE = 14 +PRODUCT_NAME_SIZE = 24 @dataclass(frozen=True) @@ -19,6 +20,8 @@ class ComplexFixture: metadata_offsets: set[int] file_id_manufacturer_offset: int file_id_product_offset: int + file_id_product_name_offset: int + file_id_product_name_size: int creator_manufacturer_offset: int creator_product_offset: int preserved_ranges: tuple[tuple[int, int], ...] @@ -45,13 +48,24 @@ def _build_complex_fixture() -> ComplexFixture: def pos() -> int: return HEADER_SIZE + len(records) - # 1. normal file_id definition/data pair. - records.extend(definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)])) + # 1. normal file_id definition/data pair, including the product_name string + # field (8) so the string write path -- which zero-fills the whole declared + # field, the riskiest byte-preservation behavior in the patcher -- is covered + # by the byte-preservation proof and not only by the patching tests. + records.extend( + definition( + 0, + FILE_ID_MESG_NUM, + [(1, 2, 0x84), (2, 2, 0x84), (8, PRODUCT_NAME_SIZE, 0x07)], + ) + ) file_id_data_start = pos() - records.extend(data(0, struct.pack(" ComplexFixture: metadata_offsets: set[int] = set() metadata_offsets.update(range(file_id_manufacturer_offset, file_id_manufacturer_offset + 2)) metadata_offsets.update(range(file_id_product_offset, file_id_product_offset + 2)) + metadata_offsets.update( + range(file_id_product_name_offset, file_id_product_name_offset + PRODUCT_NAME_SIZE) + ) metadata_offsets.update(range(creator_manufacturer_offset, creator_manufacturer_offset + 2)) metadata_offsets.update(range(creator_product_offset, creator_product_offset + 2)) @@ -96,6 +113,8 @@ def _build_complex_fixture() -> ComplexFixture: metadata_offsets=metadata_offsets, file_id_manufacturer_offset=file_id_manufacturer_offset, file_id_product_offset=file_id_product_offset, + file_id_product_name_offset=file_id_product_name_offset, + file_id_product_name_size=PRODUCT_NAME_SIZE, creator_manufacturer_offset=creator_manufacturer_offset, creator_product_offset=creator_product_offset, preserved_ranges=( @@ -152,7 +171,7 @@ def test_complex_fixture_patches_targets_and_preserves_advanced_records( result = convert_fit_device(source, output) assert is_fit_file(output) is True - assert result.patched_field_count == 4 + assert result.patched_field_count == 5 after = output.read_bytes() assert struct.unpack_from(" None: + """The string write path zero-fills the *entire* declared field. Prove that the + rewrite is confined to the product_name field's own bytes: the target string plus + a null terminator plus zero padding, with the surrounding record bytes untouched + (the enclosing preservation test already asserts the global changed-byte set).""" + source = tmp_path / "source.fit" + output = tmp_path / "output.fit" + source.write_bytes(complex_fit_bytes) + + convert_fit_device(source, output) + + start = _FIXTURE.file_id_product_name_offset + end = start + _FIXTURE.file_id_product_name_size + after = output.read_bytes() + + expected = b"Edge 1030 Plus\x00".ljust(_FIXTURE.file_id_product_name_size, b"\x00") + assert after[start:end] == expected + # The source deliberately padded past its null terminator with 0x2A bytes, so a + # write that overran (or under-cleared) the field would be visible here. + assert complex_fit_bytes[start:end] != expected + + values = read_device_field_values(output) + assert any( + v.global_message_num == FILE_ID_MESG_NUM + and v.field_num == 8 + and v.value == "Edge 1030 Plus" + and v.is_creator + for v in values + ) + + +def test_convert_fit_device_supports_12_byte_header(tmp_path: Path) -> None: + """12-byte headers carry no header CRC field, so conversion must succeed and + report header_crc=None while still rewriting the file CRC.""" + file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)]) + file_data = data(0, struct.pack(" None: path = tmp_path / "truncated.fit" path.write_bytes(make_fit(bytes([0x40, 0x00, 0x00])))