Merge pull request #383 from magv/master

Validate full file path in the static file handler

Author: asvetlov

aiohttp/web_urldispatcher.py

@@ -149,9 +149,13 @@ def __init__(self, name, prefix, directory, *,
             'GET', self.handle, name, expect_handler=expect_handler)
         self._prefix = prefix
         self._prefix_len = len(self._prefix)
-        self._directory = directory
+        self._directory = os.path.abspath(directory) + os.sep
         self._chunk_size = chunk_size
 
+        if not os.path.isdir(self._directory):
+            raise ValueError(
+                "No directory exists at '{}'".format(self._directory))
+
     def match(self, path):
         if not path.startswith(self._prefix):
             return None
@@ -167,13 +171,13 @@ def url(self, *, filename, query=None):
     def handle(self, request):
         resp = StreamResponse()
         filename = request.match_info['filename']
-        filepath = os.path.join(self._directory, filename)
-        if '..' in filename:
+        filepath = os.path.abspath(os.path.join(self._directory, filename))
+        if not filepath.startswith(self._directory):
             raise HTTPNotFound()
         if not os.path.exists(filepath) or not os.path.isfile(filepath):
             raise HTTPNotFound()
 
-        ct, encoding = mimetypes.guess_type(filename)
+        ct, encoding = mimetypes.guess_type(filepath)
         if not ct:
             ct = 'application/octet-stream'
         resp.content_type = ct
@@ -394,8 +398,6 @@ def add_static(self, prefix, path, *, name=None, expect_handler=None,
         :param path - folder with files
         """
         assert prefix.startswith('/')
-        assert os.path.isdir(path), 'Path does not directory %s' % path
-        path = os.path.abspath(path)
         if not prefix.endswith('/'):
             prefix += '/'
         route = StaticRoute(name, prefix, path,

tests/test_web_functional.py

@@ -287,7 +287,7 @@ def go(dirname, filename):
             resp.close()
 
         here = os.path.dirname(__file__)
-        filename = os.path.join(here, 'data.unknown_mime_type')
+        filename = 'data.unknown_mime_type'
         self.loop.run_until_complete(go(here, filename))
 
     def test_static_file_with_content_type(self):
@@ -319,7 +319,7 @@ def go(dirname, filename):
             resp.close()
 
         here = os.path.dirname(__file__)
-        filename = os.path.join(here, 'software_development_in_picture.jpg')
+        filename = 'software_development_in_picture.jpg'
         self.loop.run_until_complete(go(here, filename))
 
     def test_static_file_with_content_encoding(self):
@@ -342,9 +342,46 @@ def go(dirname, filename):
             resp.close()
 
         here = os.path.dirname(__file__)
-        filename = os.path.join(here, 'hello.txt.gz')
+        filename = 'hello.txt.gz'
         self.loop.run_until_complete(go(here, filename))
 
+    def test_static_file_directory_traversal_attack(self):
+
+        @asyncio.coroutine
+        def go(dirname, relpath):
+            self.assertTrue(os.path.isfile(os.path.join(dirname, relpath)))
+
+            app, _, url = yield from self.create_server('GET', '/static/')
+            app.router.add_static('/static', dirname)
+
+            url_relpath = url + relpath
+            resp = yield from request('GET', url_relpath, loop=self.loop)
+            self.assertEqual(404, resp.status)
+            resp.close()
+
+            url_relpath2 = url + 'dir/../' + filename
+            resp = yield from request('GET', url_relpath2, loop=self.loop)
+            self.assertEqual(404, resp.status)
+            resp.close()
+
+            url_abspath = \
+                url + os.path.abspath(os.path.join(dirname, filename))
+            resp = yield from request('GET', url_abspath, loop=self.loop)
+            self.assertEqual(404, resp.status)
+            resp.close()
+
+        here = os.path.dirname(__file__)
+        filename = '../README.rst'
+        self.loop.run_until_complete(go(here, filename))
+
+    def test_static_route_path_existence_check(self):
+        directory = os.path.dirname(__file__)
+        web.StaticRoute(None, "/", directory)
+
+        nodirectory = os.path.join(directory, "nonexistent-uPNiOEAg5d")
+        with self.assertRaises(ValueError):
+            web.StaticRoute(None, "/", nodirectory)
+
     def test_post_form_with_duplicate_keys(self):
 
         @asyncio.coroutine