Harden unicode/exif handling and sanitize generated paths

elodie/external/pyexiftool.py

@@ -337,15 +337,23 @@ def get_metadata_batch(self, filenames):
         The return value will have the format described in the
         documentation of :py:meth:`execute_json()`.
         """
-        return self.execute_json(*filenames)
+        data = self.execute_json(*filenames)
+        if isinstance(data, list):
+            return data
+        return []
 
     def get_metadata(self, filename):
         """Return meta-data for a single file.
 
         The returned dictionary has the format described in the
         documentation of :py:meth:`execute_json()`.
         """
-        return self.execute_json(filename)[0]
+        data = self.execute_json(filename)
+        if not isinstance(data, list) or len(data) == 0:
+            return None
+        if not isinstance(data[0], dict):
+            return None
+        return data[0]
 
     def get_tags_batch(self, tags, filenames):
         """Return only specified tags for the given files.
@@ -368,15 +376,21 @@ def get_tags_batch(self, tags, filenames):
                             "an iterable of strings")
         params = ["-" + t for t in tags]
         params.extend(filenames)
-        return self.execute_json(*params)
+        data = self.execute_json(*params)
+        if isinstance(data, list):
+            return data
+        return []
 
     def get_tags(self, tags, filename):
         """Return only specified tags for a single file.
 
         The returned dictionary has the format described in the
         documentation of :py:meth:`execute_json()`.
         """
-        return self.get_tags_batch(tags, [filename])[0]
+        data = self.get_tags_batch(tags, [filename])
+        if len(data) == 0:
+            return None
+        return data[0]
 
     def get_tag_batch(self, tag, filenames):
         """Extract a single tag from the given files.
@@ -390,9 +404,14 @@ def get_tag_batch(self, tag, filenames):
         non-existent tags, in the same order as ``filenames``.
         """
         data = self.get_tags_batch([tag], filenames)
+        if len(data) == 0:
+            return [None for _ in filenames]
         result = []
         for d in data:
-            d.pop("SourceFile")
+            if not isinstance(d, dict):
+                result.append(None)
+                continue
+            d.pop("SourceFile", None)
             result.append(next(iter(d.values()), None))
         return result
 
@@ -402,7 +421,10 @@ def get_tag(self, tag, filename):
         The return value is the value of the specified tag, or
         ``None`` if this tag was not found in the file.
         """
-        return self.get_tag_batch(tag, [filename])[0]
+        data = self.get_tag_batch(tag, [filename])
+        if len(data) == 0:
+            return None
+        return data[0]
 
     def set_tags_batch(self, tags, filenames):
         """Writes the values of the specified tags for the given files.

elodie/filesystem.py

@@ -45,10 +45,39 @@ def __init__(self):
         # See build failures in Python3 here.
         #  https://travis-ci.org/jmathai/elodie/builds/483012902
         self.whitespace_regex = '[ \t\n\r\f\v]+'
+        # Disallow path separators and filesystem-invalid characters in a single path component.
+        self.invalid_path_component_regex = r'[<>:"/\\|?*\x00-\x1f]'
+        self.windows_reserved_names = {
+            'CON', 'PRN', 'AUX', 'NUL',
+            'COM1', 'COM2', 'COM3', 'COM4', 'COM5', 'COM6', 'COM7', 'COM8', 'COM9',
+            'LPT1', 'LPT2', 'LPT3', 'LPT4', 'LPT5', 'LPT6', 'LPT7', 'LPT8', 'LPT9',
+        }
 
         # Instantiate a plugins object
         self.plugins = Plugins()
 
+    def sanitize_path_component(self, value):
+        """Sanitize a single folder/file path component for cross-platform safety."""
+        if value is None:
+            return value
+
+        value = re.sub(self.invalid_path_component_regex, '-', value)
+
+        if os.sep:
+            value = value.replace(os.sep, '-')
+        if os.altsep:
+            value = value.replace(os.altsep, '-')
+
+        value = value.rstrip(' .')
+        if len(value) == 0:
+            return ''
+
+        # Windows has reserved device names which cannot be used as path components.
+        stem = value.split('.', 1)[0].upper()
+        if stem in self.windows_reserved_names:
+            value = '_%s' % value
+
+        return value
     def _file_operation(self, operation_type, src, dst=None):
         """Perform file operation with dry-run support."""
         if constants.dry_run:
@@ -234,12 +263,16 @@ def get_file_name(self, metadata):
                     name,
                 )
             else:
+                this_value = self.sanitize_path_component(this_value)
                 name = re.sub(
                     '%{}'.format(part),
                     this_value,
                     name,
                 )
 
+        # Final guard to avoid unsafe separators from custom templates.
+        name = self.sanitize_path_component(name)
+
         config = load_config()
 
         if('File' in config and 'capitalization' in config['File'] and config['File']['capitalization'] == 'upper'):
@@ -385,7 +418,9 @@ def get_folder_path(self, metadata, path_parts=None):
                 part, mask = this_part
                 this_path = self.get_dynamic_path(part, mask, metadata)
                 if this_path:
-                    path.append(this_path.strip())
+                    this_path = self.sanitize_path_component(this_path).strip()
+                    if len(this_path) > 0:
+                        path.append(this_path)
                     # We break as soon as we have a value to append
                     # Else we continue for fallbacks
                     break

elodie/media/media.py

@@ -119,16 +119,19 @@ def get_coordinate(self, type='latitude'):
     def get_exiftool_attributes(self):
         """Get attributes for the media object from exiftool.
 
-        :returns: dict, or False if exiftool was not available.
+        :returns: dict, or None if exiftool metadata was unavailable.
         """
         source = self.source
 
         #Cache exif metadata results and use if already exists for media
         if(self.exif_metadata is None):
-            self.exif_metadata = ExifTool().get_metadata(source)
+            try:
+                self.exif_metadata = ExifTool().get_metadata(source)
+            except Exception:
+                self.exif_metadata = None
 
         if not self.exif_metadata:
-            return False
+            return None
 
         return self.exif_metadata
 

elodie/tests/external_pyexiftool_test.py

@@ -81,4 +81,18 @@ def test_exiftool_with_non_ascii_file():
         if os.path.exists(test_file):
             os.remove(test_file)
         if os.path.exists(test_dir):
-            os.rmdir(test_dir)
+            os.rmdir(test_dir)
+
+def test_get_metadata_returns_none_when_execute_json_fails():
+    """get_metadata() should not crash when execute_json returns None."""
+    et = ExifTool()
+    with patch.object(et, 'execute_json', return_value=None):
+        result = et.get_metadata("/tmp/test.jpg")
+        assert result is None
+
+def test_get_metadata_returns_none_when_execute_json_is_empty():
+    """get_metadata() should not crash when execute_json returns an empty list."""
+    et = ExifTool()
+    with patch.object(et, 'execute_json', return_value=[]):
+        result = et.get_metadata("/tmp/test.jpg")
+        assert result is None

elodie/tests/filesystem_test.py

@@ -251,6 +251,20 @@ def test_get_file_name_with_uppercase_and_spaces():
 
     assert file_name == helper.path_tz_fix('2015-12-05_00-59-26-plain-with-spaces-and-uppercase-123.jpg'), file_name
 
+def test_get_file_name_sanitizes_invalid_path_characters():
+    filesystem = FileSystem()
+    media = Photo(helper.get_file('with-title.jpg'))
+    metadata = media.get_metadata()
+    metadata['title'] = 'nami cc aapi / 中文部公众讲座 : 1?*'
+
+    file_name = filesystem.get_file_name(metadata)
+
+    assert '/' not in file_name, file_name
+    assert '\\' not in file_name, file_name
+    assert ':' not in file_name, file_name
+    assert '?' not in file_name, file_name
+    assert '*' not in file_name, file_name
+
 @mock.patch('elodie.config.get_config_file', return_value='%s/config.ini-filename-custom' % gettempdir())
 def test_get_file_name_custom(mock_get_config_file):
     with open(mock_get_config_file.return_value, 'w') as f:
@@ -395,6 +409,14 @@ def test_get_folder_path_with_location():
 
     assert path == os.path.join('2015-12-Dec','Sunnyvale'), path
 
+@mock.patch('elodie.filesystem.geolocation.place_name', return_value={'default': u'Bellevue/WA', 'city': u'Bellevue/WA'})
+def test_get_folder_path_sanitizes_location_separator(mock_place_name):
+    filesystem = FileSystem()
+    media = Photo(helper.get_file('with-location.jpg'))
+    path = filesystem.get_folder_path(media.get_metadata())
+
+    assert path == os.path.join('2015-12-Dec', 'Bellevue-WA'), path
+
 @mock.patch('elodie.config.get_config_file', return_value='%s/config.ini-original-with-camera-make-and-model' % gettempdir())
 def test_get_folder_path_with_camera_make_and_model(mock_get_config_file):
     with open(mock_get_config_file.return_value, 'w') as f:

elodie/tests/media/media_test.py

@@ -9,6 +9,7 @@
 import string
 import tempfile
 import time
+from unittest.mock import patch
 
 sys.path.insert(0, os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))))))
 sys.path.insert(0, os.path.abspath(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))))
@@ -18,6 +19,7 @@
 from elodie.media.media import Media
 from elodie.media.photo import Photo
 from elodie.media.video import Video
+from elodie.external.pyexiftool import ExifTool
 
 os.environ['TZ'] = 'GMT'
 
@@ -89,6 +91,19 @@ def test_get_original_name_invalid_file():
 
     assert original_name is None, original_name
 
+def test_get_original_name_when_exiftool_metadata_is_unavailable():
+    temporary_folder, folder = helper.create_working_folder()
+
+    origin = '%s/%s' % (folder, 'plain.jpg')
+    file = helper.get_file('plain.jpg')
+    shutil.copyfile(file, origin)
+
+    media = Media.get_class_by_file(origin, [Photo])
+    with patch.object(ExifTool, 'get_metadata', return_value=None):
+        original_name = media.get_original_name()
+
+    assert original_name is None, original_name
+
 def test_set_original_name_when_exists():
     temporary_folder, folder = helper.create_working_folder()