Merge pull request #338 from reef-technologies/get-upload-headers-fix

Fix FileVersion._get_upload_headers when key secret is empty

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Fix download integration tests on non-production environments
 * Add `B2_DEBUG_HTTP` env variable to enable network-level test debugging
 * Disable changelog validation temporarily
+* Fix `FileVersion._get_upload_headers` when encryption key is `None`
 
 ## [1.17.2] - 2022-06-24
 

b2sdk/encryption/types.py

@@ -17,6 +17,12 @@ class EncryptionAlgorithm(Enum):
 
     AES256 = 'AES256'
 
+    def get_length(self) -> int:
+        if self is EncryptionAlgorithm.AES256:
+            return int(256 / 8)
+
+        raise NotImplementedError()
+
 
 @unique
 class EncryptionMode(Enum):

b2sdk/file_version.py

@@ -10,6 +10,7 @@
 
 from typing import Dict, Optional, Union, Tuple, TYPE_CHECKING
 import re
+from copy import deepcopy
 
 from .encryption.setting import EncryptionSetting, EncryptionSettingFactory
 from .replication.types import ReplicationStatus
@@ -310,14 +311,23 @@ def _get_upload_headers(self) -> bytes:
         key and value. This implementation is in par with ADVANCED_HEADERS_LIMIT
         and is reasonable only for `has_large_header` method
         """
+
+        # sometimes secret is not available, but we want to calculate headers
+        # size anyway; to bypass this, we use a fake encryption setting
+        # with a fake key
+        sse = self.server_side_encryption
+        if sse and sse.key and sse.key.secret is None:
+            sse = deepcopy(sse)
+            sse.key.secret = b'*' * sse.algorithm.get_length()
+
         headers = self.api.raw_api.get_upload_file_headers(
             upload_auth_token=self.api.account_info.get_account_auth_token(),
             file_name=self.file_name,
             content_length=self.size,
             content_type=self.content_type,
             content_sha1=self.content_sha1,
             file_infos=self.file_info,
-            server_side_encryption=self.server_side_encryption,
+            server_side_encryption=sse,
             file_retention=self.file_retention,
             legal_hold=self.legal_hold,
         )

test/unit/file_version/test_file_version.py

@@ -9,7 +9,6 @@
 ######################################################################
 
 import pytest
-import os.path
 
 import apiver_deps
 from apiver_deps import B2Api
@@ -142,5 +141,33 @@ def test_delete_download_version(self):
         with pytest.raises(FileNotPresent):
             self.bucket.get_file_info_by_name(self.file_version.file_name)
 
+    def test_file_version_upload_headers(self):
+        file_version = self.file_version._clone(
+            server_side_encryption=EncryptionSetting(
+                EncryptionMode.SSE_C,
+                EncryptionAlgorithm.AES256,
+                EncryptionKey(None, None),
+            ),
+        )
+
+        assert file_version._get_upload_headers() == """
+            Authorization: auth_token_0
+            Content-Length: 7
+            X-Bz-File-Name: test_file
+            Content-Type: b2/x-auto
+            X-Bz-Content-Sha1: 0feca720e2c29dafb2c900713ba560e03b758711
+            X-Bz-Server-Side-Encryption-Customer-Algorithm: AES256
+            X-Bz-Server-Side-Encryption-Customer-Key: KioqKioqKioqKioqKioqKioqKioqKioqKioqKioqKio=
+            X-Bz-Server-Side-Encryption-Customer-Key-Md5: SaaDheEjzuynJH8eW6AEpQ==
+            X-Bz-File-Legal-Hold: off
+            X-Bz-File-Retention-Mode: None
+            X-Bz-File-Retention-Retain-Until-Timestamp: None
+        """.strip().replace(': ', '').replace(' ', '').replace('\n', '').encode('utf8')
+
+        assert not file_version.has_large_header
+
+        file_version.file_info['dummy'] = 'a' * 2000  # make metadata > 2k bytes
+        assert file_version.has_large_header
+
     # FileVersion.download tests are not here, because another test file already has all the facilities for such test
     # prepared