Download filename validation

b2/_internal/console_tool.py

@@ -106,6 +106,7 @@
     points_to_fifo,
     substitute_control_chars,
     unprintable_to_hex,
+    validate_b2_file_name_as_path,
 )
 from b2sdk.v3.exception import (
     B2Error,
@@ -1960,6 +1961,13 @@ def get_local_output_filepath(
         if not output_filepath.is_dir():
             return output_filepath
 
+        # Make sure the remote filename is safe to interpret as a local path
+        try:
+            validate_b2_file_name_as_path(str(file_request.download_version.file_name))
+        except ValueError as exc:
+            err_msg = unprintable_to_hex(f'{exc}: {output_filepath}')
+            raise CommandError(err_msg) from exc
+
         # If the output is directory, we're expected to download the file right there.
         # Normally, we overwrite the target without asking any questions, but in this case
         # user might be oblivious of the actual mistake he's about to commit.

changelog.d/+download-path-validation.fixed.md

@@ -0,0 +1 @@
+Reject unsafe remote filenames in the Download command.

changelog.d/+sdk-upgrade.infrastructure.md

@@ -0,0 +1 @@
+Bump sdk version to v2.12.0.

pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
 dependencies = [
     "argcomplete>=3.5.2,<4",
     "arrow>=1.0.2,<2.0.0",
-    "b2sdk>=2.10.3,<3",
+    "b2sdk>=2.12.0,<3",
     "docutils>=0.18.1,<0.22",
     "idna~=3.4; platform_system == 'Java'",
     "rst2ansi==0.1.5",

test/unit/console_tool/test_download_file.py

@@ -176,7 +176,7 @@ def test_download_file_by_name__to_stdout_by_alias(
 
 def test_cat__b2_uri(b2_cli, bucket, uploaded_stdout_txt, tmp_path, capfd):
     b2_cli.run(
-        ['file', 'cat', '--no-progress', f"b2://{bucket}/{uploaded_stdout_txt['fileName']}"],
+        ['file', 'cat', '--no-progress', f'b2://{bucket}/{uploaded_stdout_txt["fileName"]}'],
     )
     assert capfd.readouterr().out == uploaded_stdout_txt['content']
 
@@ -233,3 +233,53 @@ def test__download_file__threads(b2_cli, local_file, uploaded_file, tmp_path):
 
     assert output_path.read_text() == uploaded_file['content']
     assert b2_cli.console_tool.api.services.download_manager.get_thread_pool_size() == num_threads
+
+
+@pytest.mark.parametrize(
+    'remote_name',
+    [
+        '../escape.txt',
+        'foo/../bar.txt',
+    ],
+)
+def test_download_file_by_uri__directory_rejects_unsupported_server_name(
+    b2_cli, api_bucket, tmp_path, remote_name
+):
+    uploaded_file = api_bucket.upload_bytes(b'hello world', remote_name)
+    output_directory = tmp_path / 'downloads'
+    output_directory.mkdir()
+
+    b2_cli.run(
+        [
+            'file',
+            'download',
+            '--no-progress',
+            f'b2id://{uploaded_file.id_}',
+            str(output_directory),
+        ],
+        expected_status=1,
+        expected_stderr=None,
+    )
+
+    local_path = (output_directory / remote_name).resolve()
+    assert not local_path.exists()
+
+
+def test_download_file_by_uri__explicit_file_target_allows_unsupported_server_name(
+    b2_cli, api_bucket, tmp_path
+):
+    uploaded_file = api_bucket.upload_bytes(b'hello world', '../escape.txt')
+    output_path = tmp_path / 'output.txt'
+
+    b2_cli.run(
+        [
+            'file',
+            'download',
+            '--no-progress',
+            f'b2id://{uploaded_file.id_}',
+            str(output_path),
+        ],
+        expected_stderr=None,
+    )
+
+    assert output_path.read_text() == 'hello world'