espressif/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2026-04-27 19:13:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'fix/wl_safe_mode_eventual_corruption_v6.0' into 'release/v6.0'

Browse Source 
fix(fatfs): Make wl_fatfsgen.py aware of WL safe mode (v6.0)

See merge request espressif/esp-idf!45397
This commit is contained in:
Jiang Jiang Jian
2026-02-02 10:38:38 +08:00
parent ac6875ff16 5e66a7f051
commit ea5534e242
2 changed files with 122 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/fatfs/fatfsparse.py
+45 -40
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python
# SPDX-FileCopyrightText: 2022-2023 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import argparse
import os
@@ -9,7 +9,12 @@ from fatfs_utils.boot_sector import BootSector
from fatfs_utils.entry import Entry
from fatfs_utils.fat import FAT
from fatfs_utils.fatfs_state import BootSectorState
from fatfs_utils.utils import FULL_BYTE, LONG_NAMES_ENCODING, PAD_CHAR, FATDefaults, lfn_checksum, read_filesystem
from fatfs_utils.utils import FULL_BYTE
from fatfs_utils.utils import LONG_NAMES_ENCODING
from fatfs_utils.utils import PAD_CHAR
from fatfs_utils.utils import FATDefaults
from fatfs_utils.utils import lfn_checksum
from fatfs_utils.utils import read_filesystem
from wl_fatfsgen import remove_wl
@@ -33,7 +38,7 @@ def get_obj_name(obj_: dict, directory_bytes_: bytes, entry_position_: int, lfn_
for pos in range(entry_position_ - 1, -1, -1): # loop from the current entry back to the start
obj_address_: int = FATDefaults.ENTRY_SIZE * pos
entry_bytes_: bytes = directory_bytes_[obj_address_: obj_address_ + FATDefaults.ENTRY_SIZE]
entry_bytes_: bytes = directory_bytes_[obj_address_ : obj_address_ + FATDefaults.ENTRY_SIZE]
struct_ = Entry.parse_entry_long(entry_bytes_, lfn_checksum_)
if len(struct_.items()) > 0:
full_name[struct_['order']] = build_file_name(struct_['name1'], struct_['name2'], struct_['name3'])
@@ -42,11 +47,9 @@ def get_obj_name(obj_: dict, directory_bytes_: bytes, entry_position_: int, lfn_
return ''.join(map(lambda x: x[1], sorted(full_name.items()))) or obj_name_
def traverse_folder_tree(directory_bytes_: bytes,
name: str,
state_: BootSectorState,
fat_: FAT,
binary_array_: bytes) -> None:
def traverse_folder_tree(
directory_bytes_: bytes, name: str, state_: BootSectorState, fat_: FAT, binary_array_: bytes
) -> None:
os.makedirs(name)
assert len(directory_bytes_) % FATDefaults.ENTRY_SIZE == 0
@@ -56,7 +59,8 @@ def traverse_folder_tree(directory_bytes_: bytes,
obj_address_: int = FATDefaults.ENTRY_SIZE * i
try:
obj_: dict = Entry.ENTRY_FORMAT_SHORT_NAME.parse(
directory_bytes_[obj_address_: obj_address_ + FATDefaults.ENTRY_SIZE])
directory_bytes_[obj_address_ : obj_address_ + FATDefaults.ENTRY_SIZE]
)
except (construct.core.ConstError, UnicodeDecodeError, construct.core.StringError):
args.long_name_support = True
continue
@@ -64,15 +68,16 @@ def traverse_folder_tree(directory_bytes_: bytes,
if obj_['DIR_Attr'] == 0: # empty entry
continue
obj_name_: str = get_obj_name(obj_,
directory_bytes_,
entry_position_=i,
lfn_checksum_=lfn_checksum(obj_['DIR_Name'] + obj_['DIR_Name_ext']))
obj_name_: str = get_obj_name(
obj_,
directory_bytes_,
entry_position_=i,
lfn_checksum_=lfn_checksum(obj_['DIR_Name'] + obj_['DIR_Name_ext']),
)
if obj_['DIR_Attr'] == Entry.ATTR_ARCHIVE:
content_ = b''
if obj_['DIR_FileSize'] > 0:
content_ = fat_.get_chained_content(cluster_id_=Entry.get_cluster_id(obj_),
size=obj_['DIR_FileSize'])
content_ = fat_.get_chained_content(cluster_id_=Entry.get_cluster_id(obj_), size=obj_['DIR_FileSize'])
with open(os.path.join(name, obj_name_), 'wb') as new_file:
new_file.write(content_)
elif obj_['DIR_Attr'] == Entry.ATTR_DIRECTORY:
@@ -80,11 +85,13 @@ def traverse_folder_tree(directory_bytes_: bytes,
if obj_name_ in ('.', '..'):
continue
child_directory_bytes_ = fat_.get_chained_content(cluster_id_=obj_['DIR_FstClusLO'])
traverse_folder_tree(directory_bytes_=child_directory_bytes_,
name=os.path.join(name, obj_name_),
state_=state_,
fat_=fat_,
binary_array_=binary_array_)
traverse_folder_tree(
directory_bytes_=child_directory_bytes_,
name=os.path.join(name, obj_name_),
state_=state_,
fat_=fat_,
binary_array_=binary_array_,
)
def remove_wear_levelling_if_exists(fs_: bytes) -> bytes:
@@ -109,24 +116,18 @@ def remove_wear_levelling_if_exists(fs_: bytes) -> bytes:
if __name__ == '__main__':
desc = 'Tool for parsing fatfs image and extracting directory structure on host.'
argument_parser: argparse.ArgumentParser = argparse.ArgumentParser(description=desc)
argument_parser.add_argument('input_image',
help='Path to the image that will be parsed and extracted.')
argument_parser.add_argument('--long-name-support',
action='store_true',
help=argparse.SUPPRESS)
argument_parser.add_argument('input_image', help='Path to the image that will be parsed and extracted.')
argument_parser.add_argument('--long-name-support', action='store_true', help=argparse.SUPPRESS)
# ensures backward compatibility
argument_parser.add_argument('--wear-leveling',
action='store_true',
help=argparse.SUPPRESS)
argument_parser.add_argument('--wl-layer',
choices=['detect', 'enabled', 'disabled'],
default=None,
help="If detection doesn't work correctly, "
'you can force analyzer to or not to assume WL.')
argument_parser.add_argument('--verbose',
action='store_true',
help='Prints details about FAT image.')
argument_parser.add_argument('--wear-leveling', action='store_true', help=argparse.SUPPRESS)
argument_parser.add_argument(
'--wl-layer',
choices=['detect', 'enabled', 'disabled'],
default=None,
help="If detection doesn't work correctly, you can force analyzer to or not to assume WL.",
)
argument_parser.add_argument('--verbose', action='store_true', help='Prints details about FAT image.')
args = argument_parser.parse_args()
@@ -168,7 +169,11 @@ if __name__ == '__main__':
boot_dir_start_ = boot_sector_.boot_sector_state.root_directory_start
boot_dir_sectors = boot_sector_.boot_sector_state.root_dir_sectors_cnt
full_ = fs[boot_dir_start_: boot_dir_start_ + boot_dir_sectors * boot_sector_.boot_sector_state.sector_size]
traverse_folder_tree(full_,
boot_sector_.boot_sector_state.volume_label.rstrip(chr(PAD_CHAR)),
boot_sector_.boot_sector_state, fat, fs)
full_ = fs[boot_dir_start_ : boot_dir_start_ + boot_dir_sectors * boot_sector_.boot_sector_state.sector_size]
traverse_folder_tree(
full_,
boot_sector_.boot_sector_state.volume_label.rstrip(chr(PAD_CHAR)),
boot_sector_.boot_sector_state,
fat,
fs,
)
components/fatfs/wl_fatfsgen.py
+77 -60
View File
@@ -1,18 +1,18 @@
#!/usr/bin/env python
# SPDX-FileCopyrightText: 2021-2024 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2021-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from typing import Optional
from construct import Const
from construct import Int32ul
from construct import Struct
from fatfs_utils.exceptions import WLNotInitialized
from fatfs_utils.utils import crc32
from fatfs_utils.utils import FATDefaults
from fatfs_utils.utils import FULL_BYTE
from fatfs_utils.utils import UINT32_MAX
from fatfs_utils.utils import FATDefaults
from fatfs_utils.utils import crc32
from fatfs_utils.utils import generate_4bytes_random
from fatfs_utils.utils import get_args_for_partition_generator
from fatfs_utils.utils import UINT32_MAX
from fatfsgen import FATFS
@@ -22,32 +22,33 @@ def remove_wl(binary_image: bytes) -> bytes:
wl_state_size: int = WLFATFS.WL_STATE_HEADER_SIZE + WLFATFS.WL_STATE_RECORD_SIZE * total_sectors
wl_state_sectors_cnt: int = (wl_state_size + FATDefaults.WL_SECTOR_SIZE - 1) // FATDefaults.WL_SECTOR_SIZE
wl_state_total_size: int = wl_state_sectors_cnt * FATDefaults.WL_SECTOR_SIZE
wl_sectors_size: int = (wl_state_sectors_cnt
* FATDefaults.WL_SECTOR_SIZE
* WLFATFS.WL_STATE_COPY_COUNT
+ FATDefaults.WL_SECTOR_SIZE)
wl_sectors_size: int = (
wl_state_sectors_cnt * FATDefaults.WL_SECTOR_SIZE * WLFATFS.WL_STATE_COPY_COUNT + FATDefaults.WL_SECTOR_SIZE
)
correct_wl_configuration = binary_image[-wl_sectors_size:]
data_ = WLFATFS.WL_STATE_T_DATA.parse(correct_wl_configuration[:WLFATFS.WL_STATE_HEADER_SIZE])
data_ = WLFATFS.WL_STATE_T_DATA.parse(correct_wl_configuration[: WLFATFS.WL_STATE_HEADER_SIZE])
total_records = 0
# iterating over records field of the first copy of the state sector
for i in range(WLFATFS.WL_STATE_HEADER_SIZE, wl_state_total_size, WLFATFS.WL_STATE_RECORD_SIZE):
if correct_wl_configuration[i:i + WLFATFS.WL_STATE_RECORD_SIZE] != WLFATFS.WL_STATE_RECORD_SIZE * b'\xff':
if correct_wl_configuration[i : i + WLFATFS.WL_STATE_RECORD_SIZE] != WLFATFS.WL_STATE_RECORD_SIZE * b'\xff':
total_records += 1
else:
break
before_dummy = binary_image[:total_records * FATDefaults.WL_SECTOR_SIZE]
after_dummy = binary_image[total_records * FATDefaults.WL_SECTOR_SIZE + FATDefaults.WL_SECTOR_SIZE:]
before_dummy = binary_image[: total_records * FATDefaults.WL_SECTOR_SIZE]
after_dummy = binary_image[total_records * FATDefaults.WL_SECTOR_SIZE + FATDefaults.WL_SECTOR_SIZE :]
new_image: bytes = before_dummy + after_dummy
# remove wl sectors
new_image = new_image[:len(new_image) - (FATDefaults.WL_SECTOR_SIZE + 2 * wl_state_total_size)]
new_image = new_image[: len(new_image) - (FATDefaults.WL_SECTOR_SIZE + 2 * wl_state_total_size)]
# reorder to preserve original order
new_image = (new_image[-data_['move_count'] * FATDefaults.WL_SECTOR_SIZE:]
+ new_image[:-data_['move_count'] * FATDefaults.WL_SECTOR_SIZE])
new_image = (
new_image[-data_['move_count'] * FATDefaults.WL_SECTOR_SIZE :]
+ new_image[: -data_['move_count'] * FATDefaults.WL_SECTOR_SIZE]
)
return new_image
@@ -71,7 +72,7 @@ class WLFATFS:
'block_size' / Int32ul,
'version' / Int32ul,
'device_id' / Int32ul,
'reserved' / Const(28 * b'\x00')
'reserved' / Const(28 * b'\x00'),
)
WL_CONFIG_T_DATA = Struct(
@@ -82,31 +83,33 @@ class WLFATFS:
'updaterate' / Int32ul,
'wr_size' / Int32ul,
'version' / Int32ul,
'temp_buff_size' / Int32ul
'temp_buff_size' / Int32ul,
)
WL_CONFIG_T_HEADER_SIZE = 48
def __init__(self,
size: int = FATDefaults.SIZE,
sector_size: int = FATDefaults.SECTOR_SIZE,
reserved_sectors_cnt: int = FATDefaults.RESERVED_SECTORS_COUNT,
fat_tables_cnt: int = FATDefaults.FAT_TABLES_COUNT,
sectors_per_cluster: int = FATDefaults.SECTORS_PER_CLUSTER,
explicit_fat_type: int = None,
hidden_sectors: int = FATDefaults.HIDDEN_SECTORS,
long_names_enabled: bool = False,
num_heads: int = FATDefaults.NUM_HEADS,
oem_name: str = FATDefaults.OEM_NAME,
sec_per_track: int = FATDefaults.SEC_PER_TRACK,
volume_label: str = FATDefaults.VOLUME_LABEL,
file_sys_type: str = FATDefaults.FILE_SYS_TYPE,
use_default_datetime: bool = True,
version: int = FATDefaults.VERSION,
temp_buff_size: int = FATDefaults.TEMP_BUFFER_SIZE,
device_id: int = None,
root_entry_count: int = FATDefaults.ROOT_ENTRIES_COUNT,
media_type: int = FATDefaults.MEDIA_TYPE,
wl_mode: Optional[str] = None) -> None:
def __init__(
self,
size: int = FATDefaults.SIZE,
sector_size: int = FATDefaults.SECTOR_SIZE,
reserved_sectors_cnt: int = FATDefaults.RESERVED_SECTORS_COUNT,
fat_tables_cnt: int = FATDefaults.FAT_TABLES_COUNT,
sectors_per_cluster: int = FATDefaults.SECTORS_PER_CLUSTER,
explicit_fat_type: int | None = None,
hidden_sectors: int = FATDefaults.HIDDEN_SECTORS,
long_names_enabled: bool = False,
num_heads: int = FATDefaults.NUM_HEADS,
oem_name: str = FATDefaults.OEM_NAME,
sec_per_track: int = FATDefaults.SEC_PER_TRACK,
volume_label: str = FATDefaults.VOLUME_LABEL,
file_sys_type: str = FATDefaults.FILE_SYS_TYPE,
use_default_datetime: bool = True,
version: int = FATDefaults.VERSION,
temp_buff_size: int = FATDefaults.TEMP_BUFFER_SIZE,
device_id: int | None = None,
root_entry_count: int = FATDefaults.ROOT_ENTRIES_COUNT,
media_type: int = FATDefaults.MEDIA_TYPE,
wl_mode: str | None = None,
) -> None:
self._initialized = False
self._version = version
self._temp_buff_size = temp_buff_size
@@ -119,14 +122,17 @@ class WLFATFS:
# determine the number of required sectors (roundup to sector size)
self.wl_state_sectors = (self.wl_state_size + FATDefaults.WL_SECTOR_SIZE - 1) // FATDefaults.WL_SECTOR_SIZE
self.boot_sector_start = FATDefaults.WL_SECTOR_SIZE # shift by one "dummy" sector
self.fat_table_start = self.boot_sector_start + reserved_sectors_cnt * FATDefaults.WL_SECTOR_SIZE
wl_sectors = (WLFATFS.WL_DUMMY_SECTORS_COUNT + WLFATFS.WL_CFG_SECTORS_COUNT +
self.wl_state_sectors * WLFATFS.WL_STATE_COPY_COUNT)
wl_sectors = (
WLFATFS.WL_DUMMY_SECTORS_COUNT
+ WLFATFS.WL_CFG_SECTORS_COUNT
+ self.wl_state_sectors * WLFATFS.WL_STATE_COPY_COUNT
)
if self.wl_mode is not None and self.wl_mode == 'safe':
wl_sectors += WLFATFS.WL_SAFE_MODE_DUMP_SECTORS
self.boot_sector_start = FATDefaults.WL_SECTOR_SIZE # shift by one "dummy" sector
self.fat_table_start = self.boot_sector_start + reserved_sectors_cnt * FATDefaults.WL_SECTOR_SIZE
self.plain_fat_sectors = self.total_sectors - wl_sectors
self.plain_fatfs = FATFS(
explicit_fat_type=explicit_fat_type,
@@ -144,7 +150,7 @@ class WLFATFS:
sec_per_track=sec_per_track,
volume_label=volume_label,
file_sys_type=file_sys_type,
media_type=media_type
media_type=media_type,
)
self.fatfs_binary_image = self.plain_fatfs.state.binary_image
@@ -170,7 +176,7 @@ class WLFATFS:
updaterate=FATDefaults.UPDATE_RATE,
wr_size=FATDefaults.WR_SIZE,
version=self._version,
temp_buff_size=self._temp_buff_size
temp_buff_size=self._temp_buff_size,
)
)
@@ -180,14 +186,17 @@ class WLFATFS:
# adding three 4 byte zeros to align the structure
wl_config = wl_config_data + wl_config_crc + Int32ul.build(0) + Int32ul.build(0) + Int32ul.build(0)
self.fatfs_binary_image += (
wl_config + (FATDefaults.WL_SECTOR_SIZE - WLFATFS.WL_CONFIG_HEADER_SIZE) * FULL_BYTE)
self.fatfs_binary_image += wl_config + (FATDefaults.WL_SECTOR_SIZE - WLFATFS.WL_CONFIG_HEADER_SIZE) * FULL_BYTE
def _add_state_sectors(self) -> None:
wl_state_data = WLFATFS.WL_STATE_T_DATA.build(
dict(
pos=0,
max_pos=self.plain_fat_sectors + WLFATFS.WL_DUMMY_SECTORS_COUNT,
max_pos=(
self.plain_fat_sectors
+ WLFATFS.WL_DUMMY_SECTORS_COUNT
+ (WLFATFS.WL_SAFE_MODE_DUMP_SECTORS if self.wl_mode == 'safe' else 0)
),
move_count=0,
access_count=0,
max_count=FATDefaults.UPDATE_RATE,
@@ -200,10 +209,16 @@ class WLFATFS:
wl_state_crc = Int32ul.build(crc)
wl_state = wl_state_data + wl_state_crc
wl_state_sector_padding: bytes = (FATDefaults.WL_SECTOR_SIZE - WLFATFS.WL_STATE_HEADER_SIZE) * FULL_BYTE
wl_state_sector: bytes = (
wl_state_sectors: bytes = (
wl_state + wl_state_sector_padding + (self.wl_state_sectors - 1) * FATDefaults.WL_SECTOR_SIZE * FULL_BYTE
)
self.fatfs_binary_image += (WLFATFS.WL_STATE_COPY_COUNT * wl_state_sector)
# add 2 extra state-preservation sectors in 'Safe' mode
if self.wl_mode is not None and self.wl_mode == 'safe':
wl_safe_dummy_sec: bytes = WLFATFS.WL_SAFE_MODE_DUMP_SECTORS * FATDefaults.WL_SECTOR_SIZE * FULL_BYTE
self.fatfs_binary_image += wl_safe_dummy_sec
self.fatfs_binary_image += WLFATFS.WL_STATE_COPY_COUNT * wl_state_sectors
def wl_write_filesystem(self, output_path: str) -> None:
if not self._initialized:
@@ -215,15 +230,17 @@ class WLFATFS:
if __name__ == '__main__':
desc = 'Create a FAT filesystem with support for wear levelling and populate it with directory content'
args = get_args_for_partition_generator(desc, wl=True)
wl_fatfs = WLFATFS(size=args.partition_size,
sector_size=args.sector_size,
fat_tables_cnt=args.fat_count,
sectors_per_cluster=args.sectors_per_cluster,
explicit_fat_type=args.fat_type,
long_names_enabled=args.long_name_support,
use_default_datetime=args.use_default_datetime,
root_entry_count=args.root_entry_count,
wl_mode=args.wl_mode)
wl_fatfs = WLFATFS(
size=args.partition_size,
sector_size=args.sector_size,
fat_tables_cnt=args.fat_count,
sectors_per_cluster=args.sectors_per_cluster,
explicit_fat_type=args.fat_type,
long_names_enabled=args.long_name_support,
use_default_datetime=args.use_default_datetime,
root_entry_count=args.root_entry_count,
wl_mode=args.wl_mode,
)
wl_fatfs.plain_fatfs.generate(args.input_directory)
wl_fatfs.init_wl()