Merge pull request #748 from wiredfool/djpeg

Prevent shell injection

.travis.yml

@@ -14,7 +14,7 @@ python:
   - 3.4
 
 install:
-  - "sudo apt-get -qq install libfreetype6-dev liblcms2-dev python-qt4 ghostscript libffi-dev cmake"
+  - "sudo apt-get -qq install libfreetype6-dev liblcms2-dev python-qt4 ghostscript libffi-dev libjpeg-turbo-progs cmake"
   - "pip install cffi"
   - "pip install coveralls nose pyroma"
   - if [ "$TRAVIS_PYTHON_VERSION" == "2.6" ]; then pip install unittest2; fi

PIL/GifImagePlugin.py

@@ -333,13 +333,41 @@ def _save_netpbm(im, fp, filename):
     # below for information on how to enable this.
 
     import os
+    from subprocess import Popen, check_call, PIPE, CalledProcessError
+    import tempfile    
     file = im._dump()
+
     if im.mode != "RGB":
-        os.system("ppmtogif %s >%s" % (file, filename))
+        with open(filename, 'wb') as f:
+            stderr = tempfile.TemporaryFile()
+            check_call(["ppmtogif", file], stdout=f, stderr=stderr)
     else:
-        os.system("ppmquant 256 %s | ppmtogif >%s" % (file, filename))
+        with open(filename, 'wb') as f:
-    try: os.unlink(file)
+
-    except: pass
+            # Pipe ppmquant output into ppmtogif
+            # "ppmquant 256 %s | ppmtogif > %s" % (file, filename)
+            quant_cmd = ["ppmquant", "256", file]
+            togif_cmd = ["ppmtogif"]
+            stderr = tempfile.TemporaryFile()
+            quant_proc = Popen(quant_cmd, stdout=PIPE, stderr=stderr)
+            stderr = tempfile.TemporaryFile()
+            togif_proc = Popen(togif_cmd, stdin=quant_proc.stdout, stdout=f, stderr=stderr)
+            
+            # Allow ppmquant to receive SIGPIPE if ppmtogif exits
+            quant_proc.stdout.close()
+
+            retcode = quant_proc.wait()
+            if retcode:
+                raise CalledProcessError(retcode, quant_cmd)
+
+            retcode = togif_proc.wait()
+            if retcode:
+                raise CalledProcessError(retcode, togif_cmd)
+
+    try:
+        os.unlink(file)
+    except:
+        pass
 
 
 # --------------------------------------------------------------------

PIL/JpegImagePlugin.py

@@ -354,12 +354,14 @@ class JpegImageFile(ImageFile.ImageFile):
 
         # ALTERNATIVE: handle JPEGs via the IJG command line utilities
 
+        import subprocess
         import tempfile
         import os
         f, path = tempfile.mkstemp()
         os.close(f)
         if os.path.exists(self.filename):
-            os.system("djpeg '%s' >'%s'" % (self.filename, path))
+            with open(path, 'wb') as f:
+                subprocess.check_call(["djpeg", self.filename], stdout=f)
         else:
             raise ValueError("Invalid Filename")
 
@@ -602,8 +604,10 @@ def _save(im, fp, filename):
 def _save_cjpeg(im, fp, filename):
     # ALTERNATIVE: handle JPEGs via the IJG command line utilities.
     import os
-    file = im._dump()
+    import subprocess
-    os.system("cjpeg %s >%s" % (file, filename))
+    tempfile = im._dump()
+    with open(filename, 'wb') as f:
+        subprocess.check_call(["cjpeg", tempfile], stdout=f)
     try:
         os.unlink(file)
     except:

Tests/helper.py

@@ -199,4 +199,29 @@ def lena(mode="RGB", cache={}):
     # cache[mode] = im
     return im
 
+
+def command_succeeds(cmd):
+    """
+    Runs the command, which must be a list of strings. Returns True if the
+    command succeeds, or False if an OSError was raised by subprocess.Popen.
+    """
+    import os
+    import subprocess
+    with open(os.devnull, 'w') as f:
+        try:
+            subprocess.Popen(cmd, stdout=f, stderr=subprocess.STDOUT).wait()
+        except OSError:
+            return False
+    return True
+
+def djpeg_available():
+    return command_succeeds(['djpeg', '--help'])
+
+def cjpeg_available():
+    return command_succeeds(['cjpeg', '--help'])
+
+def netpbm_available():
+    return command_succeeds(["ppmquant", "--help"]) and \
+           command_succeeds(["ppmtogif", "--help"])
+
 # End of file

Tests/test_file_gif.py

@@ -1,6 +1,7 @@
-from helper import unittest, PillowTestCase, tearDownModule, lena
+from helper import unittest, PillowTestCase, tearDownModule, lena, netpbm_available
 
 from PIL import Image
+from PIL import GifImagePlugin
 
 codecs = dir(Image.core)
 
@@ -89,6 +90,22 @@ class TestFileGif(PillowTestCase):
         reloaded = roundtrip(im)[1].convert('RGB')
         self.assert_image_equal(im, reloaded)
 
+    @unittest.skipUnless(netpbm_available(), "netpbm not available")
+    def test_save_netpbm_bmp_mode(self):
+        img = Image.open(file).convert("RGB")
+
+        tempfile = self.tempfile("temp.gif")
+        GifImagePlugin._save_netpbm(img, 0, tempfile)
+        self.assert_image_similar(img, Image.open(tempfile).convert("RGB"), 0)
+
+    @unittest.skipUnless(netpbm_available(), "netpbm not available")
+    def test_save_netpbm_l_mode(self):
+        img = Image.open(file).convert("L")
+
+        tempfile = self.tempfile("temp.gif")
+        GifImagePlugin._save_netpbm(img, 0, tempfile)
+        self.assert_image_similar(img, Image.open(tempfile).convert("L"), 0)
+
 
 if __name__ == '__main__':
     unittest.main()

Tests/test_file_jpeg.py

@@ -1,10 +1,12 @@
 from helper import unittest, PillowTestCase, tearDownModule, lena, py3
+from helper import djpeg_available, cjpeg_available
 
 import random
 from io import BytesIO
 
 from PIL import Image
 from PIL import ImageFile
+from PIL import JpegImagePlugin
 
 codecs = dir(Image.core)
 
@@ -273,8 +275,23 @@ class TestFileJpeg(PillowTestCase):
                                                  qtables={0:standard_l_qtable,
                                                           1:standard_chrominance_qtable}),
                                   30)
-        
+
-        
+    @unittest.skipUnless(djpeg_available(), "djpeg not available")
+    def test_load_djpeg(self):
+        img = Image.open(test_file)
+        img.load_djpeg()
+        self.assert_image_similar(img, Image.open(test_file), 0)
+
+    @unittest.skipUnless(cjpeg_available(), "cjpeg not available")
+    def test_save_cjpeg(self):
+        img = Image.open(test_file)
+
+        tempfile = self.tempfile("temp.jpg")
+        JpegImagePlugin._save_cjpeg(img, 0, tempfile)
+        # Default save quality is 75%, so a tiny bit of difference is alright
+        self.assert_image_similar(img, Image.open(tempfile), 1)
+
+
 if __name__ == '__main__':
     unittest.main()
 

Tests/test_shell_injection.py

@@ -0,0 +1,56 @@
+from helper import unittest, PillowTestCase, tearDownModule
+from helper import djpeg_available, cjpeg_available, netpbm_available
+
+import shutil
+
+from PIL import Image, JpegImagePlugin, GifImagePlugin
+
+test_jpg = "Tests/images/lena.jpg"
+test_gif = "Tests/images/lena.gif"
+
+test_filenames = (
+    "temp_';",
+    "temp_\";",
+    "temp_'\"|",
+    "temp_'\"||",
+    "temp_'\"&&",
+)
+
+class TestShellInjection(PillowTestCase):
+
+    def assert_save_filename_check(self, src_img, save_func):
+        for filename in test_filenames:
+            dest_file = self.tempfile(filename)
+            save_func(src_img, 0, dest_file)
+            # If file can't be opened, shell injection probably occurred
+            Image.open(dest_file).load()
+
+    @unittest.skipUnless(djpeg_available(), "djpeg not available")
+    def test_load_djpeg_filename(self):
+        for filename in test_filenames:
+            src_file = self.tempfile(filename)
+            shutil.copy(test_jpg, src_file)
+
+            im = Image.open(src_file)
+            im.load_djpeg()
+
+    @unittest.skipUnless(cjpeg_available(), "cjpeg not available")
+    def test_save_cjpeg_filename(self):
+        im = Image.open(test_jpg)
+        self.assert_save_filename_check(im, JpegImagePlugin._save_cjpeg)
+
+    @unittest.skipUnless(netpbm_available(), "netpbm not available")
+    def test_save_netpbm_filename_bmp_mode(self):
+        im = Image.open(test_gif).convert("RGB")
+        self.assert_save_filename_check(im, GifImagePlugin._save_netpbm)
+
+    @unittest.skipUnless(netpbm_available(), "netpbm not available")
+    def test_save_netpbm_filename_l_mode(self):
+        im = Image.open(test_gif).convert("L")
+        self.assert_save_filename_check(im, GifImagePlugin._save_netpbm)
+
+
+if __name__ == '__main__':
+    unittest.main()
+
+# End of file