fix: safely handle Content-Disposition download filenames

tests/test_api_client/test_file_download.py

@@ -0,0 +1,196 @@
+# -*- coding: utf-8 -*-
+from pathlib import Path
+
+import pytest
+
+from xero_python.api_client import ApiClient
+from xero_python.api_client import copy_download_without_overwrite
+from xero_python.api_client.configuration import Configuration
+
+
+class FakeResponse:
+    def __init__(self, content_disposition, data=b"file contents"):
+        self.content_disposition = content_disposition
+        self.data = data
+
+    def getheader(self, name):
+        if name == "Content-Disposition":
+            return self.content_disposition
+        return None
+
+
+@pytest.fixture
+def api_client(tmp_path):
+    configuration = Configuration()
+    configuration.temp_folder_path = str(tmp_path)
+    return ApiClient(configuration=configuration)
+
+
+def deserialize_file(api_client, response):
+    return Path(api_client._ApiClient__deserialize_file(response))
+
+
+@pytest.mark.parametrize(
+    "header",
+    [
+        'attachment; filename="../outside.txt"',
+        'attachment; filename="..\\outside.txt"',
+    ],
+)
+def test_deserialize_file_keeps_traversal_within_temp_directory(
+    api_client, tmp_path, header
+):
+    path = deserialize_file(api_client, FakeResponse(header))
+
+    assert path.parent == tmp_path
+    assert path.name == "outside.txt"
+    assert path.read_bytes() == b"file contents"
+    assert not (tmp_path.parent / "outside.txt").exists()
+
+
+@pytest.mark.parametrize(
+    "header",
+    [
+        'attachment; filename="/outside.txt"',
+        'attachment; filename="C:\\outside.txt"',
+        'attachment; filename="\\\\server\\share\\outside.txt"',
+        'attachment; filename="NUL.txt"',
+        'attachment; filename="bad\x00name.txt"',
+        'attachment; filename="spoof\u202ename.txt"',
+        'attachment; filename="report.csv."',
+    ],
+)
+def test_deserialize_file_rejects_unsafe_cross_platform_names(
+    api_client, tmp_path, header
+):
+    path = deserialize_file(api_client, FakeResponse(header))
+
+    assert path.parent == tmp_path
+    assert path.name not in {
+        "outside.txt",
+        "NUL.txt",
+        "bad\x00name.txt",
+        "spoof\u202ename.txt",
+        "report.csv.",
+    }
+    assert path.read_bytes() == b"file contents"
+
+
+def test_deserialize_file_uses_content_disposition_filename(api_client, tmp_path):
+    path = deserialize_file(
+        api_client, FakeResponse('attachment; filename="report.csv"')
+    )
+
+    assert path == tmp_path / "report.csv"
+    assert path.read_bytes() == b"file contents"
+
+
+@pytest.mark.parametrize(
+    "header,expected",
+    [
+        (
+            "attachment; filename*=UTF-8''..%5Cencoded%20report.csv",
+            "encoded report.csv",
+        ),
+        ('attachment; filename="quarter; report.csv"', "quarter; report.csv"),
+    ],
+)
+def test_deserialize_file_parses_encoded_and_quoted_names(
+    api_client, tmp_path, header, expected
+):
+    path = deserialize_file(api_client, FakeResponse(header))
+
+    assert path == tmp_path / expected
+    assert path.read_bytes() == b"file contents"
+
+
+def test_deserialize_file_does_not_follow_existing_symlink(api_client, tmp_path):
+    target = tmp_path.parent / "symlink-target.txt"
+    target.write_bytes(b"do not overwrite")
+    link = tmp_path / "report.csv"
+    try:
+        link.symlink_to(target)
+    except OSError as error:
+        pytest.skip("symlinks are unavailable: {}".format(error))
+
+    path = deserialize_file(
+        api_client, FakeResponse('attachment; filename="report.csv"')
+    )
+
+    assert path != link
+    assert path.parent == tmp_path
+    assert path.read_bytes() == b"file contents"
+    assert link.is_symlink()
+    assert target.read_bytes() == b"do not overwrite"
+
+
+def test_deserialize_file_preserves_existing_regular_file(api_client, tmp_path):
+    destination = tmp_path / "report.csv"
+    destination.write_bytes(b"do not overwrite")
+
+    path = deserialize_file(
+        api_client, FakeResponse('attachment; filename="report.csv"')
+    )
+
+    assert path != destination
+    assert path.parent == tmp_path
+    assert path.read_bytes() == b"file contents"
+    assert destination.read_bytes() == b"do not overwrite"
+
+
+def test_copy_download_preserves_destination_created_after_validation(tmp_path):
+    source = tmp_path / "secure-random-file"
+    source.write_bytes(b"file contents")
+    destination = tmp_path / "report.csv"
+
+    destination.write_bytes(b"created by racer")
+    copied = copy_download_without_overwrite(str(source), str(destination))
+
+    assert not copied
+    assert source.read_bytes() == b"file contents"
+    assert destination.read_bytes() == b"created by racer"
+
+
+def test_copy_download_preserves_replacement_when_atomic_claim_fails(
+    tmp_path, monkeypatch
+):
+    source = tmp_path / "secure-random-file"
+    source.write_bytes(b"file contents")
+    destination = tmp_path / "report.csv"
+
+    def replace_name_and_fail(source_path, destination_path, follow_symlinks):
+        Path(destination_path).write_bytes(b"created by racer")
+        raise OSError("injected atomic-claim failure")
+
+    monkeypatch.setattr("xero_python.api_client.os.link", replace_name_and_fail)
+
+    copied = copy_download_without_overwrite(str(source), str(destination))
+
+    assert not copied
+    assert source.read_bytes() == b"file contents"
+    assert destination.read_bytes() == b"created by racer"
+
+
+@pytest.mark.parametrize(
+    "header",
+    [
+        'attachment; filename="unterminated',
+        "attachment; filename*=UTF-8''bad%ZZname",
+    ],
+)
+def test_deserialize_file_handles_malformed_content_disposition_safely(
+    api_client, tmp_path, header
+):
+    path = deserialize_file(api_client, FakeResponse(header))
+
+    assert path.parent == tmp_path
+    assert path.read_bytes() == b"file contents"
+
+
+def test_deserialize_file_uses_generated_filename_without_filename_parameter(
+    api_client, tmp_path
+):
+    path = deserialize_file(api_client, FakeResponse("inline"))
+
+    assert path.parent == tmp_path
+    assert path.read_bytes() == b"file contents"

xero_python/api_client/__init__.py

@@ -1,22 +1,25 @@
 # coding: utf-8
 """
-    Xero oAuth 2 identity service
+Xero oAuth 2 identity service
 
-    This specifing endpoints related to managing authentication tokens and identity for Xero API  # noqa: E501
+This specifing endpoints related to managing authentication tokens and identity for Xero API  # noqa: E501
 
-    OpenAPI spec version: 2.0.4
-    Contact: api@xero.com
-    Generated by: https://openapi-generator.tech
+OpenAPI spec version: 2.0.4
+Contact: api@xero.com
+Generated by: https://openapi-generator.tech
 """
 
-
 import datetime
 import json
 import mimetypes
+import ntpath
 import os
+import posixpath
 import re
 import tempfile
+import unicodedata
 from decimal import Decimal
+from email.message import Message
 from multiprocessing.pool import ThreadPool
 from urllib.parse import quote
 
@@ -27,6 +30,62 @@
 from xero_python.api_client.serializer import serialize
 from xero_python.exceptions import OAuth2TokenGetterError, OAuth2TokenSaverError
 
+WINDOWS_RESERVED_FILENAMES = {"CON", "PRN", "AUX", "NUL"}
+WINDOWS_RESERVED_FILENAMES.update("COM{}".format(number) for number in range(1, 10))
+WINDOWS_RESERVED_FILENAMES.update("LPT{}".format(number) for number in range(1, 10))
+
+
+def safe_download_filename(content_disposition):
+    message = Message()
+    try:
+        message["Content-Disposition"] = content_disposition
+        filename = message.get_filename()
+    except (TypeError, ValueError):
+        return None
+    if not filename or any(
+        unicodedata.category(character).startswith("C") for character in filename
+    ):
+        return None
+    if (
+        posixpath.isabs(filename)
+        or ntpath.isabs(filename)
+        or ntpath.splitdrive(filename)[0]
+    ):
+        return None
+
+    filename = re.split(r"[\\/]", filename)[-1]
+    if filename in ("", ".", "..") or filename != filename.rstrip(" ."):
+        return None
+    if any(character in '<>:"|?*' for character in filename):
+        return None
+    if filename.split(".", 1)[0].upper() in WINDOWS_RESERVED_FILENAMES:
+        return None
+    return filename
+
+
+def safe_download_path(directory, filename):
+    directory = os.path.realpath(directory)
+    path = os.path.join(directory, filename)
+    try:
+        if os.path.commonpath((directory, os.path.realpath(path))) != directory:
+            return None
+    except ValueError:
+        return None
+    return path
+
+
+def copy_download_without_overwrite(source, destination):
+    try:
+        os.link(source, destination, follow_symlinks=False)
+    except OSError:
+        return False
+
+    try:
+        os.remove(source)
+    except OSError:
+        pass
+    return True
+
 
 class ModelFinder:
     """
@@ -598,18 +657,25 @@ def __deserialize_file(self, response):
         :return: file path.
         """
         fd, path = tempfile.mkstemp(dir=self.configuration.temp_folder_path)
-        os.close(fd)
-        os.remove(path)
+
+        try:
+            with os.fdopen(fd, "wb") as file_handle:
+                file_handle.write(response.data)
+        except Exception:
+            os.remove(path)
+            raise
 
         content_disposition = response.getheader("Content-Disposition")
         if content_disposition:
-            filename = re.search(
-                r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition
-            ).group(1)
-            path = os.path.join(os.path.dirname(path), filename)
-
-        with open(path, "wb") as f:
-            f.write(response.data)
+            filename = safe_download_filename(content_disposition)
+            destination = (
+                safe_download_path(os.path.dirname(path), filename)
+                if filename
+                else None
+            )
+            if destination:
+                if copy_download_without_overwrite(path, destination):
+                    path = destination
 
         return path