#404: avoid unsafe tar file extraction (#410)

Author: mhucka

recirq/documentation_utils.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import inspect
+import io
 import os
 import re
 import tarfile
@@ -67,6 +68,14 @@ def fetch_guide_data_collection_data(base_dir=None):
 
     print("Downloading guide/data_collection data.")
     stream = urllib.request.urlopen("https://ndownloader.figshare.com/files/25541666")
-
-    with tarfile.open(fileobj=stream, mode='r|xz') as tf:
-        tf.extractall(path=base_dir)
+    # Read into a BytesIO object to make it seekable
+    stream_bytes = io.BytesIO(stream.read())
+
+    with tarfile.open(fileobj=stream_bytes, mode='r|xz') as tf:
+        for member in tf.getmembers():
+            # Ensure the path being extracted is safe.
+            if not os.path.abspath(os.path.join(base_dir, member.name)).startswith(
+                os.path.abspath(base_dir)
+            ):
+                raise ValueError(f"Encountered untrusted path {member.name}")
+            tf.extract(member, path=base_dir)

recirq/documentation_utils_test.py

@@ -12,6 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import os
+import io
+import tarfile
+from unittest import mock
+
+import pytest
 
 import recirq
 from recirq.readout_scan.tasks import ReadoutScanTask
@@ -35,6 +40,23 @@ def test_display_markdown_docstring():
 """
 
 
-def test_fetch_guide_data_collection_data(tmpdir):
-    recirq.fetch_guide_data_collection_data(base_dir=tmpdir)
-    assert os.path.exists(f'{tmpdir}/2020-02-tutorial')
+@mock.patch('urllib.request.urlopen')
+def test_fetch_guide_data_collection_data_traversal(mock_urlopen, tmpdir):
+    # Create a malicious tarball in memory.
+    malicious_tar_stream = io.BytesIO()
+    with tarfile.open(fileobj=malicious_tar_stream, mode='w:xz') as tf:
+        # Add a file that tries to write outside the target directory
+        malicious_info = tarfile.TarInfo(name="../../tmp/pwned")
+        tf.addfile(malicious_info, io.BytesIO(b"pwned"))
+    malicious_tar_stream.seek(0)
+
+    # Read the stream into a BytesIO object so that the mock should return a
+    # response object whose read() method returns the tarball content.
+    mock_response = mock.Mock()
+    mock_response.read.return_value = malicious_tar_stream.getvalue()
+    mock_urlopen.return_value = mock_response
+
+    with pytest.raises(ValueError, match="Encountered untrusted path"):
+        recirq.fetch_guide_data_collection_data(base_dir=tmpdir)
+
+    assert not os.path.exists('/tmp/pwned')