Improve problematic filenames support

libmat2/exiftool.py

@@ -5,7 +5,7 @@ import shutil
 import subprocess
 import tempfile
 
-from typing import Dict, Union, Set
+from typing import Dict, Union, Set, Callable, Any
 
 from . import abstract
 
@@ -20,27 +20,23 @@ class ExiftoolParser(abstract.AbstractParser):
     """
     meta_whitelist = set()  # type: Set[str]
 
-    @staticmethod
-    def __handle_problematic_filename(filename: str, callback) -> bytes:
-        """ This method takes a filename with a problematic name,
-        and safely applies it a `callback`."""
+    def _handle_problematic_filename(self, callback: Callable[[str], Any]) -> bytes:
+        """ This method takes a filename with a potentially problematic name,
+        and safely applies a `callback` to it.
+        """
+        if re.search('^[a-z0-9/]', self.filename) is not None:
+            return callback(self.filename)
+
         tmpdirname = tempfile.mkdtemp()
         fname = os.path.join(tmpdirname, "temp_file")
-        shutil.copy(filename, fname)
+        shutil.copy(self.filename, fname)
         out = callback(fname)
         shutil.rmtree(tmpdirname)
         return out
 
     def get_meta(self) -> Dict[str, Union[str, dict]]:
-        """ There is no way to escape the leading(s) dash(es) of the current
-        self.filename to prevent parameter injections, so we need to take care
-        of this.
-        """
         fun = lambda f: subprocess.check_output([_get_exiftool_path(), '-json', f])
-        if re.search('^[a-z0-9/]', self.filename) is None:
-            out = self.__handle_problematic_filename(self.filename, fun)
-        else:
-            out = fun(self.filename)
+        out = self._handle_problematic_filename(fun)
         meta = json.loads(out.decode('utf-8'))[0]
         for key in self.meta_whitelist:
             meta.pop(key, None)

libmat2/video.py

@@ -1,5 +1,6 @@
 import os
 import subprocess
+import logging
 
 from . import exiftool
 
@@ -23,13 +24,10 @@ class AVIParser(exiftool.ExiftoolParser):
                       'SampleRate', 'AvgBytesPerSec', 'BitsPerSample',
                       'Duration', 'ImageSize', 'Megapixels'}
 
-    def remove_all(self) -> bool:
-        """
-            TODO: handle problematic filenames starting with `-` and `--`,
-            check exiftool.py
-        """
+
+    def __remove_all_internal(self, filename: str):
         cmd = [_get_ffmpeg_path(),
-               '-i', self.filename,      # input file
+               '-i', filename,           # input file
                '-y',                     # overwrite existing output file
                '-loglevel', 'panic',     # Don't show log
                '-hide_banner',           # hide the banner
@@ -40,10 +38,13 @@ class AVIParser(exiftool.ExiftoolParser):
                '-flags:v', '+bitexact',  # don't add any metadata
                '-flags:a', '+bitexact',  # don't add any metadata
                self.output_filename]
+        subprocess.check_call(cmd)
 
+    def remove_all(self) -> bool:
         try:
-            subprocess.check_call(cmd)
-        except subprocess.CalledProcessError:
+            self._handle_problematic_filename(self.__remove_all_internal)
+        except subprocess.CalledProcessError as e:
+            logging.error("Something went wrong during the processing of %s: %s", self.filename, e)
             return False
         return True
 

tests/test_libmat2.py

@@ -37,6 +37,19 @@ class TestParameterInjection(unittest.TestCase):
         self.assertEqual(meta['ModifyDate'], "2018:03:20 21:59:25")
         os.remove('-ver')
 
+    def test_ffmpeg_injection(self):
+        try:
+            video._get_ffmpeg_path()
+        except RuntimeError:
+            raise unittest.SkipTest
+
+        shutil.copy('./tests/data/dirty.avi', './--output')
+        p = video.AVIParser('--output')
+        meta = p.get_meta()
+        print(meta)
+        self.assertEqual(meta['Software'], 'MEncoder SVN-r33148-4.0.1')
+        os.remove('--output')
+
 
 class TestUnsupportedEmbeddedFiles(unittest.TestCase):
     def test_odt_with_svg(self):