bastian/mywhoosh2garmin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden FIT patcher error boundary and verify patched metadata

Browse Source 
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>
This commit is contained in:
Bastian Wagner
2026-08-15 14:29:40 +02:00
parent 95aa8507f0
commit ef9ca9eeca
5 changed files with 343 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
app/fit/__init__.py
View File

@@ -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",
]

4
app/fit/models.py
View File

@@ -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

134
app/fit/rewriter.py
View File

@@ -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("<I", data, 4)[0]
offset = header_size
@@ -162,6 +185,7 @@ def convert_fit_device(
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)
@@ -170,45 +194,59 @@ def convert_fit_device(
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 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,
}:
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:
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:
@@ -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

129
tests/fit/test_rewriter_patching.py
View File

@@ -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("<HH", 255, 999))
device_def = definition(1, DEVICE_INFO_MESG_NUM, [(0, 0, 0x02), (2, 2, 0x84), (4, 2, 0x84)])
sensor = data(1, struct.pack("<HH", 32, 1234))
return make_fit(file_def + file_data + device_def + sensor)
def _build_undersized_serial_fixture() -> 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("<HHH", 255, 999, 4242))
return make_fit(file_def + file_data)
def _build_undersized_product_fixture() -> 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("<HB", 255, 99))
return make_fit(file_def + file_data)
def _build_undersized_product_name_fixture() -> 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("<HH", 255, 999) + b"MyWhoosh\x00\x00")
return make_fit(file_def + file_data)
def _read_device_info_records(path: Path) -> 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)

82
tests/fit/test_rewriter_preservation.py
View File

@@ -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("<HH", 255, 999)))
original_product_name = b"MyWhoosh Simulator\x00".ljust(PRODUCT_NAME_SIZE, b"\x2A")
records.extend(data(0, struct.pack("<HH", 255, 999) + original_product_name))
file_id_manufacturer_offset = file_id_data_start + 1 # +1 for the record header byte
file_id_product_offset = file_id_manufacturer_offset + 2 # manufacturer is a u16
file_id_product_name_offset = file_id_product_offset + 2 # product is a u16
# 2. device_info definition with a creator record (device_index == 0).
records.extend(definition(1, DEVICE_INFO_MESG_NUM, [(0, 1, 0x02), (2, 2, 0x84), (4, 2, 0x84)]))
@@ -88,6 +102,9 @@ def _build_complex_fixture() -> 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("<H", after, _FIXTURE.file_id_manufacturer_offset)[0] == 1
@@ -172,6 +191,61 @@ def test_complex_fixture_patches_targets_and_preserves_advanced_records(
)
def test_product_name_string_write_stays_inside_its_declared_field(
tmp_path: Path, complex_fit_bytes: bytes
) -> 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("<HH", 255, 999))
source = tmp_path / "source12.fit"
source.write_bytes(make_fit(file_def + file_data, header_size=12))
output = tmp_path / "output12.fit"
result = convert_fit_device(source, output)
assert result.header_crc is None
assert result.file_crc is not None
assert result.patched_field_count == 2
assert is_fit_file(output) is True
assert output.read_bytes()[0] == 12
values = {(v.global_message_num, v.field_num): v.value for v in read_device_field_values(output)}
assert values[(FILE_ID_MESG_NUM, 1)] == 1
assert values[(FILE_ID_MESG_NUM, 2)] == 3570
def test_truncated_definition_is_non_recoverable(tmp_path: Path) -> None:
path = tmp_path / "truncated.fit"
path.write_bytes(make_fit(bytes([0x40, 0x00, 0x00])))