Load imperfect images and allow non-standard file names (#11, #12)

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):

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

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
 

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