diff --git a/dfsimage/cli.py b/dfsimage/cli.py index fcbcc95..0154751 100644 --- a/dfsimage/cli.py +++ b/dfsimage/cli.py @@ -431,6 +431,8 @@ def __init__(self, namespace): self.verbose = (getattr(namespace, "verbose", 0) + 1 - getattr(namespace, "quiet", 0)) self.silent = getattr(namespace, "silent", None) + if getattr(namespace, "allow_invalid_names", False): + Image.allow_invalid_names = True @staticmethod def get_sector_number(image, sector): @@ -1335,6 +1337,9 @@ def _add_global_options(parser, subparser=True, template=False): action=argparse.BooleanOptionalAction, default=True) # pylint: disable=no-member add('-s', '--silent', action='store_true', help="Don't generate error if a file doesn't exist.") + add('--allow-invalid-names', action='store_true', + help="Allow non-standard characters (e.g. control codes) in file names " + "of created files, with a warning instead of an error.") def _drecat_command(namespace, _parser): diff --git a/dfsimage/entry.py b/dfsimage/entry.py index 9e5635b..76a2f77 100644 --- a/dfsimage/entry.py +++ b/dfsimage/entry.py @@ -74,6 +74,51 @@ def _isnamechar(char: int) -> bool: """ return 32 <= char < 127 + @staticmethod + def _isnamechar_lax(char: int) -> bool: + """Check if character can be stored in a catalog entry at all. + + File and directory names are stored as 7-bit values, so any code + in range 0 - 127 can be stored. Codes outside of the standard range + checked by :meth:`_isnamechar` are used on some disks, typically as + a form of copy protection. + + Args: + c: ASCII code of the character. + Returns: + A boolean indicating if the character can be stored. + """ + return 0 <= char < 128 + + @staticmethod + def _escape_name(name: Union[str, bytes]) -> str: + """Make file name with control characters printable for messages.""" + if isinstance(name, bytes): + name = name.decode("latin-1") + return "".join(c if 32 <= ord(c) < 127 else "\\x%02x" % ord(c) + for c in name) + + def _check_name_bytes(self, value: bytes, what: str) -> bool: + """Check characters in a file or directory name about to be stored. + + Returns: + ``True`` if the name contains non-standard characters which are + allowed by the :data:`Image.allow_invalid_names` setting, ``False`` + if all characters are standard. + Raises: + ValueError: The name contains characters that are not allowed. + """ + if all(Entry._isnamechar(x) for x in value): + return False + if (all(Entry._isnamechar_lax(x) for x in value) + and getattr(self.side.image, "allow_invalid_names", False)): + return True + raise ValueError("invalid %s '%s'" % (what, Entry._escape_name(value))) + + def _warn_invalid_name(self, name: bytes) -> None: + warn(ValidationWarning("Invalid characters in file name '%s'" + % Entry._escape_name(name))) + @property def directory_bytes(self) -> bytes: """bytes: File directory name as raw byte. @@ -82,8 +127,13 @@ def directory_bytes(self) -> bytes: @directory_bytes.setter def directory_bytes(self, value: bytes) -> None: - if len(value) != 1 or not Entry._isnamechar(value[0]): + if len(value) != 1: raise ValueError("invalid directory name") + if self._check_name_bytes(value, "directory name"): + self._warn_invalid_name(value + b'.' + self.filename_bytes) + self._set_directory_bytes(value) + + def _set_directory_bytes(self, value: bytes) -> None: self.side.modified = True self.entry1[7] = (self.entry1[7] & 128) | value[0] # type: ignore @@ -117,8 +167,13 @@ def filename_bytes(self) -> bytes: @filename_bytes.setter def filename_bytes(self, value: bytes) -> None: value = value.ljust(7) - if len(value) != 7 or any(not Entry._isnamechar(x) for x in value): + if len(value) != 7: raise ValueError("invalid file name") + if self._check_name_bytes(value, "file name"): + self._warn_invalid_name(self.directory_bytes + b'.' + value.rstrip(b' ')) + self._set_filename_bytes(value) + + def _set_filename_bytes(self, value: bytes) -> None: self.side.modified = True self.entry1[0:7] = value # type: ignore @@ -167,9 +222,20 @@ def fullname(self) -> str: def fullname(self, value: str) -> None: value = self.side._to_fullname(value) if len(value) > 1 and value[1] == '.': - self.directory, _, self.filename = value.partition('.') + directory, _, filename = value.partition('.') + dir_bytes = unicode_to_bbc(directory).encode("ascii") + name_bytes = unicode_to_bbc(filename).encode("ascii").ljust(7) + if len(dir_bytes) != 1 or len(name_bytes) != 7: + raise ValueError("invalid file name '%s'" % Entry._escape_name(value)) + # Check both parts before writing anything, and warn only once + invalid = self._check_name_bytes(dir_bytes, "directory name") + invalid |= self._check_name_bytes(name_bytes, "file name") + if invalid: + self._warn_invalid_name(dir_bytes + b'.' + name_bytes.rstrip(b' ')) + self._set_directory_bytes(dir_bytes) + self._set_filename_bytes(name_bytes) else: - ValueError("invalid file name '%s'" % value) + raise ValueError("invalid file name '%s'" % Entry._escape_name(value)) @property def fullname_ascii(self) -> str: @@ -690,7 +756,8 @@ def inf(self) -> str: def get_inf(self) -> 'Inf': """Create :class:`Inf` object for this file.""" inf = Inf() - inf.filename = self.fullname_ascii.lstrip() + # Strip only spaces - control characters are part of the name + inf.filename = self.fullname_ascii.lstrip(' ') inf.load_addr = self.load_address inf.exec_addr = self.exec_address inf.size = self.size diff --git a/dfsimage/image.py b/dfsimage/image.py index 94ce840..8271ec7 100644 --- a/dfsimage/image.py +++ b/dfsimage/image.py @@ -74,11 +74,23 @@ class Image: "{is_valid:1}|{sha1}" ) + #: bool: Allow non-standard characters in new file names. + #: + #: DFS file and directory names are normally limited to printable ASCII + #: characters (codes 32 - 126). Some disks, typically copy-protected + #: games, use other 7-bit codes, e.g. control characters. Such disks are + #: opened with a validation warning, but by default creating or renaming + #: a file with such a name is an error. Set this to ``True`` (on the class + #: to change the default, or on an instance) to issue a warning instead. + #: Character codes 128 and above are always rejected. + allow_invalid_names: bool = False + def __init__(self, filename: str, for_write: bool = False, open_mode: OpenMode = None, heads: int = None, tracks: int = None, linear: bool = None, warn_mode: WarnMode = None, index: Union[int, MMBEntry] = None, - catalog_only=False) -> None: + catalog_only=False, *, + allow_invalid_names: Optional[bool] = None) -> None: """Open disk image file and construct a new :class:`Image` object. Args: @@ -101,6 +113,8 @@ def __init__(self, filename: str, for_write: bool = False, catalog_only (bool): Open image only for reading catalog data. Only catalog sectors are read from disk and attempt to access any data outside catalog will raise exception. + allow_invalid_names (Optional[bool]): Allow non-standard characters + in file names. See :data:`allow_invalid_names`. Raises: RuntimeError: Image file is invalid or the class doesn't like it for some reason. @@ -113,6 +127,9 @@ def __init__(self, filename: str, for_write: bool = False, """ self._modified = False + if allow_invalid_names is not None: + self.allow_invalid_names = allow_invalid_names + filename, index = self._parse_index(filename, index) #: Image file name self.filename: str = os.path.basename(filename) @@ -137,6 +154,9 @@ def __init__(self, filename: str, for_write: bool = False, #: Linear double-sided image. Always ``True`` for single-sides images. self.linear: int = 0 self._original_size = 0 + # Actual size of the image file. May differ from _original_size + # if the last sector in the file is incomplete. + self._file_size = 0 self._second_catalog_offset = 0 self._offset = 0 self._mmb_file: Optional[MMBFileProtocol] = None @@ -226,7 +246,12 @@ def _check_mmb_file(self, open_mode: OpenMode, and not os.path.exists(self.path)) if not self.is_new_image: - self._original_size = os.path.getsize(self.path) + self._file_size = os.path.getsize(self.path) + # Some images found in the wild have the last sector + # incomplete. Treat the image as if the last sector was + # padded to full size - a warning is issued in _load_image. + self._original_size = (-(-self._file_size // SECTOR_SIZE) + * SECTOR_SIZE) mmb_count = is_mmb_file(self.path) @@ -272,6 +297,7 @@ def _get_image_format(self, heads: Optional[int], self.linear = True self._original_size = MMB_DISK_SIZE + self._file_size = MMB_DISK_SIZE else: # Default to single side unless file extension is 'dsd' or image is @@ -294,7 +320,7 @@ def _peek_number_of_tracks(self) -> int: if self.is_new_image: return DOUBLE_TRACKS - if self._original_size < CATALOG_SECTORS * SECTOR_SIZE: + if self._file_size < CATALOG_SECTORS * SECTOR_SIZE: raise RuntimeError("%s: disk image too small" % self.filename) catalog_data = bytearray(CATALOG_SECTORS * SECTOR_SIZE) @@ -318,12 +344,8 @@ def _validate_image_parameters(self, raise ValueError("invalid index value") if not self.is_new_image: - # Sanity check - image file size should be multiple of sector size - if self._original_size % SECTOR_SIZE != 0: - raise RuntimeError("%s: invalid disk image size" % self.filename) - # Make sure that at least first side catalog sectors are present - if self._original_size < CATALOG_SECTORS * SECTOR_SIZE: + if self._file_size < CATALOG_SECTORS * SECTOR_SIZE: raise RuntimeError("%s: disk image too small" % self.filename) # Sanity check @@ -383,24 +405,62 @@ def _load_image(self, warn_mode: Optional[WarnMode], open_mode: OpenMode): b_count = self.file.readinto(self._dataview) # type: ignore[attr-defined] - if b_count != self._original_size and b_count < len(self._dataview): + if b_count != self._file_size and b_count < len(self._dataview): raise RuntimeError("%s: unexpected image short read" % self.filename) # Validate the image self.validate(warn_mode) - # Sanity check. Validate the image first to know how to - # calculate min_size - if self._original_size < self.min_size: - raise RuntimeError("%s: disk image too small" % self.filename) + # Images found in the wild are sometimes shorter than they + # should be. Missing data reads as empty (0xE5) sectors, + # which is what emulators do, so warn rather than fail. + self._check_image_size(warn_mode) except: # noqa: E722 self.close(False) raise + def _files_end(self) -> int: + """Offset of the end of the last sector occupied by any file. + + Unlike :data:`min_size`, this is computed from the catalog entries + even if the catalog is invalid, so it can be used to check whether + all file data is present in an image file. Sector numbers past + the physical end of the disk side are clipped. + """ + end = 0 + for side in self.sides: + if side is None: + continue + last = CATALOG_SECTORS + total = side.total_sectors + for entry in side.files: + start = min(entry.start_sector, total) + last = max(last, min(max(entry.end_sector, start), total)) + end = max(end, self._logical_sector_end(side.head, last - 1)) + return end + + def _check_image_size(self, warn_mode: Optional[WarnMode]) -> None: + """Issue warnings if the image file is incomplete.""" + if self.is_mmb or self.catalog_only or warn_mode == WarnMode.NONE: + return + + if self._file_size != self._original_size: + warn(ValidationWarning( + "%s: disk image size (%d) is not a multiple of sector size, " + "last sector padded" % (self.displayname, self._file_size))) + + files_end = self._files_end() + if self._original_size < files_end: + warn(ValidationWarning( + "%s: disk image truncated - files extend to offset %d, " + "but image size is %d; missing data reads as empty sectors" + % (self.displayname, files_end, self._file_size))) + @classmethod def create(cls, filename: str, heads: int = None, tracks: int = None, - linear: bool = None, index: Union[int, MMBEntry] = None) -> 'Image': + linear: bool = None, index: Union[int, MMBEntry] = None, *, + allow_invalid_names: Optional[bool] = None) -> 'Image': """Create new image file. The `Image` object created by this function keeps an open file handle to the @@ -429,18 +489,21 @@ def create(cls, filename: str, heads: int = None, tracks: int = None, index (Optional[Union[int, MMBEntry]): Can be used to select default side for the created Image object. Valid values are 0 for first side and 1 for second side. + allow_invalid_names (Optional[bool]): Allow non-standard characters + in file names. See :data:`allow_invalid_names`. Raises: ValueError: The ``heads`` or ``tracks`` argument has invalid value. Returns: New :class:`Image` object. """ - return cls(filename, True, OpenMode.NEW, heads, tracks, linear, None, index) + return cls(filename, True, OpenMode.NEW, heads, tracks, linear, None, index, + allow_invalid_names=allow_invalid_names) @classmethod def open(cls, filename: str, for_write=False, open_mode: OpenMode = None, heads: int = None, tracks: int = None, linear: bool = None, warn_mode: WarnMode = None, index: Union[int, MMBEntry] = None, - catalog_only=False) -> 'Image': + catalog_only=False, *, allow_invalid_names: Optional[bool] = None) -> 'Image': """Open disk image file. The `Image` object created by this function keeps an open file handle to the @@ -475,6 +538,8 @@ def open(cls, filename: str, for_write=False, open_mode: OpenMode = None, catalog_only (bool): Open image only for reading catalog data. Only catalog sectors are read from disk and attempt to access any data outside catalog will raise exception. + allow_invalid_names (Optional[bool]): Allow non-standard characters + in file names. See :data:`allow_invalid_names`. Raises: ValueError: The ``heads`` or ``tracks`` argument has invalid value. ValueError: The ``open_mode`` argument is invalid or ``open_mode`` @@ -487,7 +552,8 @@ def open(cls, filename: str, for_write=False, open_mode: OpenMode = None, """ return cls(filename, for_write, open_mode, heads, tracks, linear, - warn_mode, index, catalog_only) + warn_mode, index, catalog_only, + allow_invalid_names=allow_invalid_names) def save(self, size_option: SizeOption = None) -> None: """Write image data back to file. @@ -1601,8 +1667,9 @@ def _to_fullname(self, filename: str, if head is None: head = self._default_head - # remove trailing spaces - filename = filename.rstrip() + # remove trailing spaces (only spaces - other whitespace characters + # may be a part of a non-standard file name) + filename = filename.rstrip(' ') # get drive from filename filename, f_dir, f_head = self._parse_name(filename, False) @@ -1625,9 +1692,14 @@ def _to_fullname(self, filename: str, filename = "%s.%s" % (f_dir, filename) # validate filename characters - if any(not Entry._isnamechar(ord(c)) - for c in filename): - raise ValueError("invalid characters in filename '%s'" % filename) + if any(not Entry._isnamechar(ord(c)) for c in unicode_to_bbc(filename)): + if not self.allow_invalid_names: + raise ValueError("invalid characters in filename '%s' " + "(use --allow-invalid-names to allow)" + % Entry._escape_name(filename)) + if any(not Entry._isnamechar_lax(ord(c)) for c in unicode_to_bbc(filename)): + raise ValueError("invalid characters in filename '%s'" + % Entry._escape_name(filename)) return filename, head diff --git a/dfsimage/inf.py b/dfsimage/inf.py index 7dec00a..c9d033f 100644 --- a/dfsimage/inf.py +++ b/dfsimage/inf.py @@ -124,7 +124,9 @@ def _partition(cls, value: str, allow_spaces: bool) -> Tuple[str, List[str]]: if name_end == -1: raise ValueError("invalid inf file") - name = value[:name_end].rstrip() + # Strip only spaces - other whitespace characters (e.g. form feed) + # may be a part of the DFS file name + name = value[:name_end].rstrip(' ') # Empty name - i.e. inf line starts with space if len(name) == 0: @@ -356,7 +358,7 @@ def scandir(self) -> None: # Skip inf file if there is no matching data file if (not entry.is_file() or not entry.name.lower().endswith(".inf") - or not os.path.exists(entry.name[:-4])): + or not os.path.exists(entry.path[:-4])): continue # Convert to lower case if possible diff --git a/dfsimage/protocol.py b/dfsimage/protocol.py index 084e697..99c93e6 100644 --- a/dfsimage/protocol.py +++ b/dfsimage/protocol.py @@ -39,6 +39,7 @@ class ImageProtocol(Protocol): default_side: int locked: bool initialized: bool + allow_invalid_names: bool _mmb_entry: Optional[MMBEntry] MMB_STATUS_MAP: Dict[int, str] diff --git a/readme.rst b/readme.rst index f13e55a..689f468 100644 --- a/readme.rst +++ b/readme.rst @@ -247,6 +247,10 @@ global options Don't generate error if a file doesn't exist. ``--continue, --no-continue`` Continue on non-fatal errors. (default: True) +``--allow-invalid-names`` + Allow non-standard characters (e.g. control codes) in file names of created + files, with a warning instead of an error. Such names are found on some + copy-protected disks, so this option allows them to be rebuilt. common command options ------------------------ diff --git a/tests/test_tolerant_images.py b/tests/test_tolerant_images.py new file mode 100644 index 0000000..ac5ad28 --- /dev/null +++ b/tests/test_tolerant_images.py @@ -0,0 +1,206 @@ +# pylint: disable=redefined-outer-name, unused-argument +# pylint: disable=missing-module-docstring, missing-function-docstring +"""Tests for loading imperfect images (#11) and non-standard names (#12).""" +import io +import os + +import pytest + +import dfsimage +from dfsimage import simplewarn + +SECTOR = 256 + + +@pytest.fixture +def warnings(monkeypatch): + """Capture warnings printed by simplewarn.""" + output = io.StringIO() + monkeypatch.setattr(simplewarn, "stderr", output) + return output + + +def make_image(path, files, tracks=80): + with dfsimage.Image.create(path, tracks=tracks) as image: + for name, data in files: + image.add_file(name, data, load_addr=0x1900, exec_addr=0x8023) + return path + + +def truncate(path, size): + with open(path, "r+b") as file: + file.truncate(size) + + +def patch(path, offset, data): + with open(path, "r+b") as file: + file.seek(offset) + file.write(data) + + +# --- Issue #11: images that are shorter than they should be ------------- + +def test_partial_last_sector_is_padded(tmpdir, warnings): + # Like Return to Eden / Dracula's Castle: the last sector is incomplete, + # but all file data is present. + path = make_image(os.path.join(tmpdir, "odd.ssd"), + [("GAME", b"G" * 1000), ("!BOOT", b"CHAIN\"GAME\"\r")]) + truncate(path, 4 * SECTOR + 1000 - 48) + + with dfsimage.Image.open(path) as image: + assert image.isvalid + assert image.get_entry("GAME").readall()[:952] == b"G" * 952 + + assert "not a multiple of sector size" in warnings.getvalue() + assert "truncated" not in warnings.getvalue() + + +def test_partial_last_sector_saved_as_full_sector(tmpdir, warnings): + path = make_image(os.path.join(tmpdir, "odd.ssd"), [("GAME", b"G" * 1000)]) + truncate(path, 3 * SECTOR + 100) + + with dfsimage.Image.open(path, for_write=True) as image: + image.sides[0].title = "FIXED" + + assert os.path.getsize(path) % SECTOR == 0 + with dfsimage.Image.open(path) as image: + assert image.sides[0].title == "FIXED" + + +def test_non_standard_sector_count_with_short_image(tmpdir, warnings): + # Like Dominium (293 sectors) / Boulder Dash Compendium (320 sectors): + # the catalog is technically invalid, which used to make dfsimage demand + # a full-size image, even though all file data is present. + path = make_image(os.path.join(tmpdir, "odd_sectors.ssd"), + [("GAME", b"D" * 2000)]) + patch(path, SECTOR + 6, bytes([0x31, 0x25])) # boot opt 3, 0x125 sectors + truncate(path, 0x125 * SECTOR) + + with dfsimage.Image.open(path) as image: + assert not image.isvalid + assert image.get_entry("GAME").readall() == b"D" * 2000 + + output = warnings.getvalue() + assert "Invalid total number of sectors (293)" in output + assert "truncated" not in output + + +def test_file_data_missing_reads_as_empty(tmpdir, warnings): + path = make_image(os.path.join(tmpdir, "short.ssd"), [("GAME", b"T" * 5000)]) + truncate(path, 2 * SECTOR + 2048) + + with dfsimage.Image.open(path) as image: + data = image.get_entry("GAME").readall() + assert data[:2048] == b"T" * 2048 + assert data[2048:] == b"\xe5" * (5000 - 2048) + + assert "truncated" in warnings.getvalue() + + +def test_warn_none_silences_size_warnings(tmpdir, warnings): + path = make_image(os.path.join(tmpdir, "short.ssd"), [("GAME", b"T" * 5000)]) + truncate(path, 2 * SECTOR + 2000) + + with dfsimage.Image.open(path, warn_mode=dfsimage.WarnMode.NONE): + pass + + assert warnings.getvalue() == "" + + +def test_missing_catalog_is_still_an_error(tmpdir): + path = make_image(os.path.join(tmpdir, "tiny.ssd"), []) + truncate(path, SECTOR + 10) + + with pytest.raises(RuntimeError, match="too small"): + dfsimage.Image.open(path) + + +# --- Issue #12: non-standard characters in file names ------------------- + +NAMES = ["P.\x0c\x15M", "\x18.\x18", "$.\x05", "$.\x7fDEL", "$.A\x00B"] + + +def make_names_image(path): + with dfsimage.Image.create(path, allow_invalid_names=True) as image: + for index, name in enumerate(NAMES): + image.add_file(name, bytes([index]) * 100, load_addr=0x1900 + index, + locked=index % 2 == 0) + image.add_file("$.NORMAL", b"N" * 300, load_addr=0xE00) + return path + + +def snapshot(image): + return sorted((entry.fullname_ascii, entry.load_address, entry.exec_address, + entry.locked, bytes(entry.readall())) + for entry in image.files) + + +def test_invalid_name_is_error_by_default(tmpdir): + with dfsimage.Image.create(os.path.join(tmpdir, "new.ssd")) as image: + with pytest.raises(ValueError, match="allow-invalid-names"): + image.add_file("$.\x05", b"X") + + +def test_invalid_name_is_warning_when_allowed(tmpdir, warnings): + path = os.path.join(tmpdir, "new.ssd") + with dfsimage.Image.create(path, allow_invalid_names=True) as image: + image.add_file("P.\x0c\x15M", b"X") + + assert warnings.getvalue().count("Invalid characters in file name") == 1 + assert "P.\\x0c\\x15M" in warnings.getvalue() + + with dfsimage.Image.open(path) as image: + assert image.files.__next__().fullname_ascii == "P.\x0c\x15M" + + +def test_top_bit_characters_always_rejected(tmpdir): + with dfsimage.Image.create(os.path.join(tmpdir, "new.ssd"), + allow_invalid_names=True) as image: + with pytest.raises(ValueError): + image.add_file("$.\x85BAD", b"X") + + +def test_round_trip_through_inf_files(tmpdir, warnings, monkeypatch): + src_path = make_names_image(os.path.join(tmpdir, "src.ssd")) + outdir = os.path.join(tmpdir, "files") + os.mkdir(outdir) + # Run from a different directory - inf files must be found regardless + monkeypatch.chdir(tmpdir) + + with dfsimage.Image.open(src_path) as src: + original = snapshot(src) + src.export_files(outdir) + + data_files = [os.path.join(outdir, name) for name in os.listdir(outdir) + if not name.endswith(".inf")] + + with dfsimage.Image.create(os.path.join(tmpdir, "strict.ssd")) as dst: + with pytest.raises(ValueError): + dst.import_files(data_files, continue_on_error=False) + + dst_path = os.path.join(tmpdir, "rebuilt.ssd") + with dfsimage.Image.create(dst_path, allow_invalid_names=True) as dst: + assert dst.import_files(data_files, continue_on_error=False) == len(original) + + with dfsimage.Image.open(dst_path) as dst: + assert snapshot(dst) == original + + +def test_copy_over_and_rename(tmpdir, warnings): + src_path = make_names_image(os.path.join(tmpdir, "src.ssd")) + dst_path = os.path.join(tmpdir, "copy.ssd") + + with dfsimage.Image.open(src_path) as src, \ + dfsimage.Image.create(dst_path, allow_invalid_names=True) as dst: + dst.copy_over(src, "*.*", continue_on_error=False) + dst.rename("$.NORMAL", "$.N\x01RMAL") + + with dfsimage.Image.open(dst_path) as dst: + names = {entry.fullname_ascii for entry in dst.files} + assert names == set(NAMES) | {"$.N\x01RMAL"} + + +def test_class_default(tmpdir, warnings, monkeypatch): + monkeypatch.setattr(dfsimage.Image, "allow_invalid_names", True) + with dfsimage.Image.create(os.path.join(tmpdir, "new.ssd")) as image: + image.add_file("$.\x05", b"X")