diff --git a/tests/test_api_client/test_file_download.py b/tests/test_api_client/test_file_download.py new file mode 100644 index 00000000..7e1fa28d --- /dev/null +++ b/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" diff --git a/xero_python/api_client/__init__.py b/xero_python/api_client/__init__.py index 645cf506..f94da764 100644 --- a/xero_python/api_client/__init__.py +++ b/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