From 63727e890f9b6a563e1fef96cdd71870d9e898b9 Mon Sep 17 00:00:00 2001 From: Kordian Kowalski Date: Wed, 2 Sep 2026 20:06:59 +0200 Subject: [PATCH 1/2] Bucket default encryption is SSE-B2; "none" is no longer a valid bucket default --- b2sdk/_internal/encryption/setting.py | 11 + b2sdk/_internal/encryption/types.py | 2 +- b2sdk/_internal/exception.py | 9 + b2sdk/_internal/raw_api.py | 48 ++-- b2sdk/_internal/raw_simulator.py | 50 +++- b2sdk/v3/exception.py | 2 + .../+file-write-encryption-check.added.md | 1 + changelog.d/+sse-b2-default.changed.md | 1 + test/integration/test_encryption.py | 236 ++++++++++++++++++ test/integration/test_raw_api.py | 29 ++- test/unit/api/test_api.py | 90 ++++--- test/unit/bucket/test_bucket.py | 214 ++++++++++++++-- test/unit/replication/test_monitoring.py | 18 +- test/unit/test_raw_api.py | 171 +++++++++++++ test/unit/test_raw_simulator.py | 74 ++++++ test/unit/v0/test_bucket.py | 53 ++-- test/unit/v0/test_raw_api.py | 6 + test/unit/v1/test_raw_api.py | 6 + 18 files changed, 897 insertions(+), 124 deletions(-) create mode 100644 changelog.d/+file-write-encryption-check.added.md create mode 100644 changelog.d/+sse-b2-default.changed.md create mode 100644 test/integration/test_encryption.py create mode 100644 test/unit/test_raw_api.py diff --git a/b2sdk/_internal/encryption/setting.py b/b2sdk/_internal/encryption/setting.py index 07663d7c4..31506cbf4 100644 --- a/b2sdk/_internal/encryption/setting.py +++ b/b2sdk/_internal/encryption/setting.py @@ -128,6 +128,17 @@ def serialize_to_json_for_request(self): raise ValueError('cannot use an unknown key in requests') return self.as_dict() + def can_be_used_for_file_write(self) -> bool: + if self.mode == EncryptionMode.SSE_B2: + return self.algorithm == EncryptionAlgorithm.AES256 and self.key is None + if self.mode == EncryptionMode.SSE_C: + return ( + self.algorithm == EncryptionAlgorithm.AES256 + and self.key is not None + and self.key.secret is not None + ) + return False + def as_dict(self): """ Represent the setting as a dict, for example: diff --git a/b2sdk/_internal/encryption/types.py b/b2sdk/_internal/encryption/types.py index 811c3f5e7..5848b3271 100644 --- a/b2sdk/_internal/encryption/types.py +++ b/b2sdk/_internal/encryption/types.py @@ -42,4 +42,4 @@ def can_be_set_as_bucket_default(self): ENCRYPTION_MODES_WITH_MANDATORY_ALGORITHM = frozenset((EncryptionMode.SSE_B2, EncryptionMode.SSE_C)) ENCRYPTION_MODES_WITH_MANDATORY_KEY = frozenset((EncryptionMode.SSE_C,)) -BUCKET_DEFAULT_ENCRYPTION_MODES = frozenset((EncryptionMode.NONE, EncryptionMode.SSE_B2)) +BUCKET_DEFAULT_ENCRYPTION_MODES = frozenset((EncryptionMode.SSE_B2,)) diff --git a/b2sdk/_internal/exception.py b/b2sdk/_internal/exception.py index 11a574753..aef410f37 100644 --- a/b2sdk/_internal/exception.py +++ b/b2sdk/_internal/exception.py @@ -536,6 +536,15 @@ def __str__(self): return f'{self.encryption_mode} cannot be used as default for a bucket.' +class WrongEncryptionSettingForFileWrite(InvalidUserInput): + def __init__(self, encryption_setting): + super().__init__() + self.encryption_setting = encryption_setting + + def __str__(self): + return f'{self.encryption_setting} cannot be used for file writes.' + + class CopyArgumentsMismatch(InvalidUserInput): pass diff --git a/b2sdk/_internal/raw_api.py b/b2sdk/_internal/raw_api.py index 639fc91a7..b24e5ba31 100644 --- a/b2sdk/_internal/raw_api.py +++ b/b2sdk/_internal/raw_api.py @@ -33,6 +33,7 @@ SSECKeyError, UnusableFileName, WrongEncryptionModeForBucketDefault, + WrongEncryptionSettingForFileWrite, ) from .file_lock import BucketRetentionSetting, FileRetentionSetting, LegalHold from .http_constants import FILE_INFO_HEADER_PREFIX @@ -449,11 +450,6 @@ def get_upload_file_headers( for k, v in file_info.items(): headers[FILE_INFO_HEADER_PREFIX + k] = b2_url_encode(v) if server_side_encryption is not None: - assert server_side_encryption.mode in ( - EncryptionMode.NONE, - EncryptionMode.SSE_B2, - EncryptionMode.SSE_C, - ) server_side_encryption.add_to_upload_headers(headers) if legal_hold is not None: @@ -870,11 +866,8 @@ def start_large_file( ): kwargs = {} if server_side_encryption is not None: - assert server_side_encryption.mode in ( - EncryptionMode.NONE, - EncryptionMode.SSE_B2, - EncryptionMode.SSE_C, - ) + if not server_side_encryption.can_be_used_for_file_write(): + raise WrongEncryptionSettingForFileWrite(server_side_encryption) kwargs['serverSideEncryption'] = server_side_encryption.serialize_to_json_for_request() if server_side_encryption.mode == EncryptionMode.SSE_C: @@ -1057,6 +1050,11 @@ def upload_file( :param custom_upload_timestamp: custom upload timestamp for the file :return: """ + if ( + server_side_encryption is not None + and not server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(server_side_encryption) # Raise UnusableFileName if the file_name doesn't meet the rules. self.check_b2_filename(file_name) headers = self.get_upload_file_headers( @@ -1088,6 +1086,11 @@ def upload_part( data_stream, server_side_encryption: EncryptionSetting | None = None, ): + if ( + server_side_encryption is not None + and not server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(server_side_encryption) headers = { 'Authorization': upload_auth_token, 'Content-Length': str(content_length), @@ -1095,11 +1098,6 @@ def upload_part( 'X-Bz-Content-Sha1': content_sha1, } if server_side_encryption is not None: - assert server_side_encryption.mode in ( - EncryptionMode.NONE, - EncryptionMode.SSE_B2, - EncryptionMode.SSE_C, - ) server_side_encryption.add_to_upload_headers(headers) return self.b2_http.post_content_return_json( @@ -1125,6 +1123,11 @@ def copy_file( file_retention: FileRetentionSetting | None = None, legal_hold: LegalHold | None = None, ): + if ( + destination_server_side_encryption is not None + and not destination_server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(destination_server_side_encryption) kwargs = {} if bytes_range is not None: range_dict = {} @@ -1152,11 +1155,6 @@ def copy_file( if destination_bucket_id is not None: kwargs['destinationBucketId'] = destination_bucket_id if destination_server_side_encryption is not None: - assert destination_server_side_encryption.mode in ( - EncryptionMode.NONE, - EncryptionMode.SSE_B2, - EncryptionMode.SSE_C, - ) kwargs['destinationServerSideEncryption'] = ( destination_server_side_encryption.serialize_to_json_for_request() ) @@ -1195,17 +1193,17 @@ def copy_part( destination_server_side_encryption: EncryptionSetting | None = None, source_server_side_encryption: EncryptionSetting | None = None, ): + if ( + destination_server_side_encryption is not None + and not destination_server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(destination_server_side_encryption) kwargs = {} if bytes_range is not None: range_dict = {} _add_range_header(range_dict, bytes_range) kwargs['range'] = range_dict['Range'] if destination_server_side_encryption is not None: - assert destination_server_side_encryption.mode in ( - EncryptionMode.NONE, - EncryptionMode.SSE_B2, - EncryptionMode.SSE_C, - ) kwargs['destinationServerSideEncryption'] = ( destination_server_side_encryption.serialize_to_json_for_request() ) diff --git a/b2sdk/_internal/raw_simulator.py b/b2sdk/_internal/raw_simulator.py index 291366422..450f7bec3 100644 --- a/b2sdk/_internal/raw_simulator.py +++ b/b2sdk/_internal/raw_simulator.py @@ -23,7 +23,7 @@ from requests.structures import CaseInsensitiveDict from .b2http import ResponseContextManager -from .encryption.setting import EncryptionMode, EncryptionSetting +from .encryption.setting import SSE_B2_AES, EncryptionMode, EncryptionSetting from .exception import ( AccessDenied, BadJson, @@ -46,6 +46,8 @@ SSECKeyError, Unauthorized, UnsatisfiableRange, + WrongEncryptionModeForBucketDefault, + WrongEncryptionSettingForFileWrite, ) from .file_lock import ( NO_RETENTION_BUCKET_SETTING, @@ -589,7 +591,7 @@ def __init__( self.file_id_to_file: dict[str, FileSimulator] = dict() self.file_name_and_id_to_file: dict[tuple[str, str], FileSimulator] = dict() if default_server_side_encryption is None: - default_server_side_encryption = EncryptionSetting(mode=EncryptionMode.NONE) + default_server_side_encryption = SSE_B2_AES self.default_server_side_encryption = default_server_side_encryption self.is_file_lock_enabled = is_file_lock_enabled self.default_retention = NO_RETENTION_BUCKET_SETTING @@ -883,7 +885,9 @@ def copy_file( ) destination_bucket = self.api.bucket_id_to_bucket.get(destination_bucket_id, self) - sse = destination_server_side_encryption or self.default_server_side_encryption + sse = ( + destination_server_side_encryption or destination_bucket.default_server_side_encryption + ) copy_file_sim = self.FILE_SIMULATOR_CLASS( self.account_id, destination_bucket, @@ -1450,6 +1454,11 @@ def create_bucket( is_file_lock_enabled: bool | None = None, replication: ReplicationConfiguration | None = None, ): + if ( + default_server_side_encryption is not None + and not default_server_side_encryption.mode.can_be_set_as_bucket_default() + ): + raise WrongEncryptionModeForBucketDefault(default_server_side_encryption.mode) if not re.match(r'^[-a-zA-Z0-9]*$', bucket_name): raise BadJson('illegal bucket name: ' + bucket_name) self._assert_account_auth(api_url, account_auth_token, account_id, 'writeBuckets') @@ -1698,6 +1707,11 @@ def copy_file( file_retention: FileRetentionSetting | None = None, legal_hold: LegalHold | None = None, ): + if ( + destination_server_side_encryption is not None + and not destination_server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(destination_server_side_encryption) bucket_id = self.file_id_to_bucket_id[source_file_id] bucket = self._get_bucket_by_id(bucket_id) self._assert_account_auth(api_url, account_auth_token, bucket.account_id, 'writeFiles') @@ -1737,6 +1751,11 @@ def copy_part( destination_server_side_encryption: EncryptionSetting | None = None, source_server_side_encryption: EncryptionSetting | None = None, ): + if ( + destination_server_side_encryption is not None + and not destination_server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(destination_server_side_encryption) if ( destination_server_side_encryption is not None and destination_server_side_encryption.mode == EncryptionMode.SSE_B2 @@ -1916,6 +1935,11 @@ def start_large_file( legal_hold: LegalHold | None = None, custom_upload_timestamp: int | None = None, ): + if ( + server_side_encryption is not None + and not server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(server_side_encryption) bucket = self._get_bucket_by_id(bucket_id) self._assert_account_auth(api_url, account_auth_token, bucket.account_id, 'writeFiles') result = bucket.start_large_file( @@ -1948,6 +1972,11 @@ def update_bucket( replication: ReplicationConfiguration | None = None, is_file_lock_enabled: bool | None = None, ): + if ( + default_server_side_encryption is not None + and not default_server_side_encryption.mode.can_be_set_as_bucket_default() + ): + raise WrongEncryptionModeForBucketDefault(default_server_side_encryption.mode) assert ( bucket_type or bucket_info @@ -2021,6 +2050,11 @@ def upload_file( legal_hold: LegalHold | None = None, custom_upload_timestamp: int | None = None, ): + if ( + server_side_encryption is not None + and not server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(server_side_encryption) with ConcurrentUsedAuthTokenGuard( self.currently_used_auth_tokens[upload_auth_token], upload_auth_token ): @@ -2033,11 +2067,6 @@ def upload_file( bucket_id, upload_id = url_match.groups() bucket = self._get_bucket_by_id(bucket_id) if server_side_encryption is not None: - assert server_side_encryption.mode in ( - EncryptionMode.NONE, - EncryptionMode.SSE_B2, - EncryptionMode.SSE_C, - ) file_info = server_side_encryption.add_key_id_to_file_info(file_info) # we don't really need headers further on @@ -2084,6 +2113,11 @@ def upload_part( input_stream, server_side_encryption: EncryptionSetting | None = None, ): + if ( + server_side_encryption is not None + and not server_side_encryption.can_be_used_for_file_write() + ): + raise WrongEncryptionSettingForFileWrite(server_side_encryption) with ConcurrentUsedAuthTokenGuard( self.currently_used_auth_tokens[upload_auth_token], upload_auth_token ): diff --git a/b2sdk/v3/exception.py b/b2sdk/v3/exception.py index 061fc385e..36143d567 100644 --- a/b2sdk/v3/exception.py +++ b/b2sdk/v3/exception.py @@ -83,6 +83,7 @@ UnsatisfiableRange, UnusableFileName, WrongEncryptionModeForBucketDefault, + WrongEncryptionSettingForFileWrite, interpret_b2_error, ) from b2sdk._internal.scan.exception import EmptyDirectory @@ -177,4 +178,5 @@ 'SSECKeyIdMismatchInCopy', 'SSECKeyError', 'WrongEncryptionModeForBucketDefault', + 'WrongEncryptionSettingForFileWrite', ) diff --git a/changelog.d/+file-write-encryption-check.added.md b/changelog.d/+file-write-encryption-check.added.md new file mode 100644 index 000000000..70f5ce242 --- /dev/null +++ b/changelog.d/+file-write-encryption-check.added.md @@ -0,0 +1 @@ +Add `EncryptionSetting.can_be_used_for_file_write()` and the `WrongEncryptionSettingForFileWrite` exception. diff --git a/changelog.d/+sse-b2-default.changed.md b/changelog.d/+sse-b2-default.changed.md new file mode 100644 index 000000000..99e6de602 --- /dev/null +++ b/changelog.d/+sse-b2-default.changed.md @@ -0,0 +1 @@ +Follow B2's switch to SSE-B2 as the default server-side encryption: an omitted encryption setting now means the server default (SSE-B2) for both bucket defaults and file writes, and an explicit `none` is rejected before the request is sent, with `WrongEncryptionModeForBucketDefault` for bucket create/update and the new `WrongEncryptionSettingForFileWrite` for uploads and copies. `RawSimulator` applies the same rules. diff --git a/test/integration/test_encryption.py b/test/integration/test_encryption.py new file mode 100644 index 000000000..182f277fa --- /dev/null +++ b/test/integration/test_encryption.py @@ -0,0 +1,236 @@ +###################################################################### +# +# File: test/integration/test_encryption.py +# +# Copyright 2026 Backblaze Inc. All Rights Reserved. +# +# License https://www.backblaze.com/using_b2_code.html +# +###################################################################### +from __future__ import annotations + +import io + +import pytest + +from b2sdk.v3 import ( + SSE_B2_AES, + EncryptionAlgorithm, + EncryptionKey, + EncryptionMode, + EncryptionSetting, + UploadSourceBytes, + WriteIntent, +) +from b2sdk.v3.exception import ( + WrongEncryptionModeForBucketDefault, + WrongEncryptionSettingForFileWrite, +) + +NO_ENCRYPTION = EncryptionSetting(mode=EncryptionMode.NONE) +UNKNOWN_ENCRYPTION = EncryptionSetting(mode=EncryptionMode.UNKNOWN) +SSE_C_AES = EncryptionSetting( + mode=EncryptionMode.SSE_C, + algorithm=EncryptionAlgorithm.AES256, + key=EncryptionKey(secret=b'customer-key', key_id='customer-key-id'), +) +DEFAULT_ENCRYPTION = SSE_B2_AES +ENCRYPTION_OMITTED = object() +FILE_CONTENTS = b'hello world' +INVALID_BUCKET_ENCRYPTIONS = [ + pytest.param(NO_ENCRYPTION, id='no-encryption'), + pytest.param(UNKNOWN_ENCRYPTION, id='unknown'), + pytest.param(SSE_C_AES, id='sse-c'), +] +INVALID_FILE_ENCRYPTIONS = [ + pytest.param(NO_ENCRYPTION, id='no-encryption'), + pytest.param(UNKNOWN_ENCRYPTION, id='unknown'), +] + + +def encryption_kwargs(encryption): + if encryption is ENCRYPTION_OMITTED: + return {} + return {'encryption': encryption} + + +@pytest.fixture(scope='module') +def multipart_data(b2_api): + part_size = b2_api.account_info.get_absolute_minimum_part_size() + 1 + return b'x' * (part_size * 2), part_size + + +class TestBucketDefaultEncryption: + @pytest.mark.parametrize( + 'encryption', + [ + pytest.param(ENCRYPTION_OMITTED, id='omitted'), + pytest.param(SSE_B2_AES, id='explicit-sse-b2'), + ], + ) + def test_create_bucket_is_sse_b2(self, b2_api, bucket_manager, encryption): + kwargs = ( + {} + if encryption is ENCRYPTION_OMITTED + else {'default_server_side_encryption': encryption} + ) + bucket = b2_api.create_bucket(bucket_manager.new_bucket_name(), 'allPrivate', **kwargs) + try: + assert bucket.default_server_side_encryption == DEFAULT_ENCRYPTION + finally: + bucket_manager.clean_bucket(bucket) + + @pytest.mark.parametrize('encryption', INVALID_BUCKET_ENCRYPTIONS) + def test_create_bucket_rejects_invalid_encryption(self, b2_api, bucket_manager, encryption): + with pytest.raises(WrongEncryptionModeForBucketDefault): + b2_api.create_bucket( + bucket_manager.new_bucket_name(), + 'allPrivate', + default_server_side_encryption=encryption, + ) + + @pytest.mark.parametrize('encryption', INVALID_BUCKET_ENCRYPTIONS) + def test_update_bucket_rejects_invalid_encryption(self, bucket, encryption): + with pytest.raises(WrongEncryptionModeForBucketDefault): + bucket.update(default_server_side_encryption=encryption) + + +class FileCreationEncryptionTestBase: + def _create_file(self, bucket, tmp_path, encryption): + raise NotImplementedError + + def test_defaults_to_sse_b2(self, bucket, tmp_path): + file_version = self._create_file(bucket, tmp_path, ENCRYPTION_OMITTED) + + assert file_version.server_side_encryption == DEFAULT_ENCRYPTION + + def test_accepts_explicit_sse_b2(self, bucket, tmp_path): + file_version = self._create_file(bucket, tmp_path, SSE_B2_AES) + + assert file_version.server_side_encryption == SSE_B2_AES + + @pytest.mark.parametrize('encryption', INVALID_FILE_ENCRYPTIONS) + def test_rejects_invalid_encryption(self, bucket, tmp_path, encryption): + with pytest.raises(WrongEncryptionSettingForFileWrite): + self._create_file(bucket, tmp_path, encryption) + + +class TestUploadBytesEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + return bucket.upload_bytes( + FILE_CONTENTS, + 'file', + **encryption_kwargs(encryption), + ) + + +class TestUploadLocalFileEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + source = tmp_path / 'source' + source.write_bytes(FILE_CONTENTS) + return bucket.upload_local_file( + source, + 'file', + **encryption_kwargs(encryption), + ) + + +class TestUploadUnboundStreamEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + return bucket.upload_unbound_stream( + io.BytesIO(FILE_CONTENTS), + 'file', + **encryption_kwargs(encryption), + ) + + +class TestUploadEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + return bucket.upload( + UploadSourceBytes(FILE_CONTENTS), + 'file', + **encryption_kwargs(encryption), + ) + + +class TestCreateFileEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + return bucket.create_file( + [WriteIntent(UploadSourceBytes(FILE_CONTENTS))], + 'file', + **encryption_kwargs(encryption), + ) + + +class TestCreateFileStreamEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + return bucket.create_file_stream( + iter([WriteIntent(UploadSourceBytes(FILE_CONTENTS))]), + 'file', + **encryption_kwargs(encryption), + ) + + +class TestConcatenateEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + return bucket.concatenate( + [UploadSourceBytes(FILE_CONTENTS)], + 'file', + **encryption_kwargs(encryption), + ) + + +class TestConcatenateStreamEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + return bucket.concatenate_stream( + iter([UploadSourceBytes(FILE_CONTENTS)]), + 'file', + **encryption_kwargs(encryption), + ) + + +class TestCopyEncryption(FileCreationEncryptionTestBase): + def _create_file(self, bucket, tmp_path, encryption): + source = bucket.upload_bytes(FILE_CONTENTS, 'source') + kwargs = {} if encryption is ENCRYPTION_OMITTED else {'destination_encryption': encryption} + return bucket.copy( + source.id_, + 'file', + **kwargs, + ) + + +class TestMultipartUploadEncryption(FileCreationEncryptionTestBase): + @pytest.fixture(autouse=True) + def _multipart_data(self, multipart_data): + self.data, self.part_size = multipart_data + + def _create_file(self, bucket, tmp_path, encryption): + file_version = bucket.upload_unbound_stream( + io.BytesIO(self.data), + 'file', + recommended_upload_part_size=self.part_size, + **encryption_kwargs(encryption), + ) + assert file_version._type() == 'large' + return file_version + + +class TestMultipartCopyEncryption(FileCreationEncryptionTestBase): + @pytest.fixture(autouse=True) + def _multipart_data(self, multipart_data): + self.data, self.part_size = multipart_data + + def _create_file(self, bucket, tmp_path, encryption): + source = bucket.upload_bytes(self.data, 'source') + kwargs = {} if encryption is ENCRYPTION_OMITTED else {'destination_encryption': encryption} + file_version = bucket.copy( + source.id_, + 'file', + length=len(self.data), + min_part_size=self.part_size, + max_part_size=self.part_size, + **kwargs, + ) + assert file_version._type() == 'large' + return file_version diff --git a/test/integration/test_raw_api.py b/test/integration/test_raw_api.py index ef3376c2e..41ace9f99 100644 --- a/test/integration/test_raw_api.py +++ b/test/integration/test_raw_api.py @@ -22,10 +22,15 @@ from b2sdk._internal.b2http import B2Http from b2sdk._internal.encryption.setting import ( EncryptionAlgorithm, + EncryptionKey, EncryptionMode, EncryptionSetting, ) -from b2sdk._internal.exception import DisablingFileLockNotSupported, Unauthorized +from b2sdk._internal.exception import ( + DisablingFileLockNotSupported, + Unauthorized, + WrongEncryptionModeForBucketDefault, +) from b2sdk._internal.file_lock import ( NO_RETENTION_FILE_SETTING, BucketRetentionSetting, @@ -335,10 +340,9 @@ def raw_api_test_helper(raw_api, should_cleanup_old_buckets): mode=EncryptionMode.SSE_B2, algorithm=EncryptionAlgorithm.AES256, ) - sse_none = EncryptionSetting(mode=EncryptionMode.NONE) for encryption_setting, default_retention in [ ( - sse_none, + None, BucketRetentionSetting(mode=RetentionMode.GOVERNANCE, period=RetentionPeriod(days=1)), ), (sse_b2_aes, None), @@ -354,6 +358,25 @@ def raw_api_test_helper(raw_api, should_cleanup_old_buckets): default_retention=default_retention, ) + for invalid_encryption_setting in [ + EncryptionSetting(mode=EncryptionMode.NONE), + EncryptionSetting(mode=EncryptionMode.UNKNOWN), + EncryptionSetting( + mode=EncryptionMode.SSE_C, + algorithm=EncryptionAlgorithm.AES256, + key=EncryptionKey(secret=b'********************************', key_id='some-id'), + ), + ]: + with pytest.raises(WrongEncryptionModeForBucketDefault): + raw_api.update_bucket( + api_url, + account_auth_token, + account_id, + bucket_id, + 'allPublic', + default_server_side_encryption=invalid_encryption_setting, + ) + # b2_list_buckets print('b2_list_buckets') bucket_list_dict = raw_api.list_buckets(api_url, account_auth_token, account_id) diff --git a/test/unit/api/test_api.py b/test/unit/api/test_api.py index d3127f84f..99bc75526 100644 --- a/test/unit/api/test_api.py +++ b/test/unit/api/test_api.py @@ -34,7 +34,13 @@ RawSimulator, RetentionMode, ) -from apiver_deps_exception import AccessDenied, FileNotPresent, InvalidArgument, RestrictedBucket +from apiver_deps_exception import ( + AccessDenied, + FileNotPresent, + InvalidArgument, + RestrictedBucket, + WrongEncryptionModeForBucketDefault, +) from ..test_base import create_key, create_key_multibucket @@ -105,7 +111,7 @@ def test_get_file_info(self): 'fileName': 'file', 'fileRetention': {'isClientAuthorizedToRead': True, 'value': {'mode': None}}, 'legalHold': {'isClientAuthorizedToRead': True, 'value': None}, - 'serverSideEncryption': {'mode': 'none'}, + 'serverSideEncryption': {'algorithm': 'AES256', 'mode': 'SSE-B2'}, 'uploadTimestamp': 5000, } else: @@ -125,7 +131,7 @@ def test_get_file_info_by_name(self): 'fileId': '9999', 'fileName': 'file', 'fileInfo': {}, - 'serverSideEncryption': {'mode': 'none'}, + 'serverSideEncryption': {'algorithm': 'AES256', 'mode': 'SSE-B2'}, 'legalHold': None, 'fileRetention': {'mode': None, 'retainUntilTimestamp': None}, 'size': 11, @@ -227,7 +233,7 @@ def test_get_file_info_by_name_with_properties(self): 'revision': 1, 'defaultServerSideEncryption': { 'isClientAuthorizedToRead': True, - 'value': {'mode': 'none'}, + 'value': {'algorithm': 'AES256', 'mode': 'SSE-B2'}, }, 'fileLockConfiguration': { 'isClientAuthorizedToRead': True, @@ -276,44 +282,65 @@ def list_buckets(*args, **kwargs): assert list_buckets(bucket_id='ID-2', use_cache=True) == [] assert self.api.list_buckets() == [] - def test_buckets_with_encryption(self): + def test_bucket_encryption_explicit_and_default(self): self._authorize_account() sse_b2_aes = EncryptionSetting( mode=EncryptionMode.SSE_B2, algorithm=EncryptionAlgorithm.AES256, ) - no_encryption = EncryptionSetting( - mode=EncryptionMode.NONE, - ) - unknown_encryption = EncryptionSetting( - mode=EncryptionMode.UNKNOWN, - ) b1 = self.api.create_bucket( 'bucket1', 'allPrivate', default_server_side_encryption=sse_b2_aes, ) - self._verify_if_bucket_is_encrypted(b1, should_be_encrypted=True) + self._assert_bucket_is_sse_b2(b1) b2 = self.api.create_bucket('bucket2', 'allPrivate') - self._verify_if_bucket_is_encrypted(b2, should_be_encrypted=False) + self._assert_bucket_is_sse_b2(b2) # uses list_buckets - self._check_if_bucket_is_encrypted('bucket1', should_be_encrypted=True) - self._check_if_bucket_is_encrypted('bucket2', should_be_encrypted=False) + self._assert_listed_bucket_is_sse_b2('bucket1') + self._assert_listed_bucket_is_sse_b2('bucket2') - # update to set encryption on b2 + # Explicitly setting SSE-B2 during an update is supported. b2.update(default_server_side_encryption=sse_b2_aes) - self._check_if_bucket_is_encrypted('bucket1', should_be_encrypted=True) - self._check_if_bucket_is_encrypted('bucket2', should_be_encrypted=True) + self._assert_listed_bucket_is_sse_b2('bucket2') - # update to unset encryption again - b2.update(default_server_side_encryption=no_encryption) - self._check_if_bucket_is_encrypted('bucket1', should_be_encrypted=True) - self._check_if_bucket_is_encrypted('bucket2', should_be_encrypted=False) + def test_bucket_rejects_plaintext_default(self): + self._authorize_account() + no_encryption = EncryptionSetting(mode=EncryptionMode.NONE) + + with pytest.raises(WrongEncryptionModeForBucketDefault): + self.api.create_bucket( + 'bucket1', + 'allPrivate', + default_server_side_encryption=no_encryption, + ) + + bucket = self.api.create_bucket('bucket2', 'allPrivate') + + with pytest.raises(WrongEncryptionModeForBucketDefault): + bucket.update(default_server_side_encryption=no_encryption) + + assert [bucket.name for bucket in self.api.list_buckets()] == ['bucket2'] + self._assert_listed_bucket_is_sse_b2('bucket2') + + def test_bucket_encryption_is_unknown_without_read_permission(self): + self._authorize_account() + sse_b2_aes = EncryptionSetting( + mode=EncryptionMode.SSE_B2, + algorithm=EncryptionAlgorithm.AES256, + ) + unknown_encryption = EncryptionSetting(mode=EncryptionMode.UNKNOWN) + + self.api.create_bucket( + 'bucket1', + 'allPrivate', + default_server_side_encryption=sse_b2_aes, + ) + self.api.create_bucket('bucket2', 'allPrivate') - # now check it with no readBucketEncryption permission to see that it's unknown key = create_key(self.api, ['listBuckets'], 'key1') self.api.authorize_account( application_key_id=key.id_, @@ -329,25 +356,16 @@ def test_buckets_with_encryption(self): assert buckets['bucket2'].default_server_side_encryption == unknown_encryption - def _check_if_bucket_is_encrypted(self, bucket_name, should_be_encrypted): + def _assert_listed_bucket_is_sse_b2(self, bucket_name): buckets = {b.name: b for b in self.api.list_buckets()} bucket = buckets[bucket_name] - return self._verify_if_bucket_is_encrypted(bucket, should_be_encrypted) + self._assert_bucket_is_sse_b2(bucket) - def _verify_if_bucket_is_encrypted(self, bucket, should_be_encrypted): - sse_b2_aes = EncryptionSetting( + def _assert_bucket_is_sse_b2(self, bucket): + assert bucket.default_server_side_encryption == EncryptionSetting( mode=EncryptionMode.SSE_B2, algorithm=EncryptionAlgorithm.AES256, ) - no_encryption = EncryptionSetting( - mode=EncryptionMode.NONE, - ) - if not should_be_encrypted: - assert bucket.default_server_side_encryption == no_encryption - else: - assert bucket.default_server_side_encryption == sse_b2_aes - assert bucket.default_server_side_encryption.mode == EncryptionMode.SSE_B2 - assert bucket.default_server_side_encryption.algorithm == EncryptionAlgorithm.AES256 def test_list_buckets_with_id(self): self._authorize_account() diff --git a/test/unit/bucket/test_bucket.py b/test/unit/bucket/test_bucket.py index a6a3f6656..7633b66d1 100644 --- a/test/unit/bucket/test_bucket.py +++ b/test/unit/bucket/test_bucket.py @@ -12,6 +12,7 @@ import contextlib import dataclasses import datetime +import inspect import io import os import pathlib @@ -46,6 +47,7 @@ SourceReplicationConflict, SSECKeyError, UnsatisfiableRange, + WrongEncryptionSettingForFileWrite, ) from test.helpers import NonSeekableIO, assert_dict_equal_ignore_extra @@ -65,7 +67,6 @@ LARGE_FILE_SHA1, NO_RETENTION_FILE_SETTING, SSE_B2_AES, - SSE_NONE, AbstractDownloader, AbstractProgressListener, B2Api, @@ -115,6 +116,7 @@ algorithm=EncryptionAlgorithm.AES256, key=EncryptionKey(secret=b'some_key', key_id='some-id'), ) +SSE_NONE = EncryptionSetting(mode=EncryptionMode.NONE) SSE_C_AES_NO_SECRET = EncryptionSetting( mode=EncryptionMode.SSE_C, algorithm=EncryptionAlgorithm.AES256, @@ -330,6 +332,31 @@ def setUp(self): self.bucket = self.api.create_bucket(self.bucket_name, 'allPublic') self.bucket_id = self.bucket.id_ + @contextlib.contextmanager + def _record_raw_api_calls(self, method_names): + """ + Record every call the SDK makes to the given RawSimulator methods + as (method name, bound arguments), then let the call through. + """ + calls = [] + + def recorder_for(name): + original = getattr(self.simulator, name) + signature = inspect.signature(original) + + def recorder(*args, **kwargs): + bound = signature.bind(*args, **kwargs) + bound.apply_defaults() + calls.append((name, bound.arguments)) + return original(*args, **kwargs) + + return recorder + + with contextlib.ExitStack() as stack: + for name in method_names: + stack.enter_context(mock.patch.object(self.simulator, name, recorder_for(name))) + yield calls + def bucket_ls(self, *args, show_versions=False, **kwargs): return bucket_ls(self.bucket, *args, show_versions=show_versions, **kwargs) @@ -501,7 +528,15 @@ def test_version_by_name(self): self.assertIsInstance(info, VFileVersionInfo) else: self.assertIsInstance(info, DownloadVersion) - expected = (a_id, 'a', 11, 'b2/x-auto', 'none', NO_RETENTION_FILE_SETTING, LegalHold.UNSET) + expected = ( + a_id, + 'a', + 11, + 'b2/x-auto', + 'SSE-B2', + NO_RETENTION_FILE_SETTING, + LegalHold.UNSET, + ) actual = ( info.id_, info.file_name, @@ -562,7 +597,7 @@ def test_version_by_id(self): info = self.bucket.get_file_info_by_id(b_id) self.assertIsInstance(info, VFileVersionInfo) - expected = (b_id, 'b', 11, 'upload', 'b2/x-auto', 'none') + expected = (b_id, 'b', 11, 'upload', 'b2/x-auto', 'SSE-B2') actual = ( info.id_, info.file_name, @@ -1228,29 +1263,25 @@ def test_encryption(self): data = b'hello world' a = self.bucket.upload_bytes(data, 'a') a_id = a.id_ - self.assertEqual(a.server_side_encryption, SSE_NONE) + self.assertEqual(a.server_side_encryption, SSE_B2_AES) b = self.bucket.upload_bytes(data, 'b', encryption=SSE_B2_AES) self.assertEqual(b.server_side_encryption, SSE_B2_AES) b_id = b.id_ - # c_id = self.bucket.upload_bytes(data, 'c', encryption=SSE_NONE).id_ # TODO self.bucket.copy(a_id, 'd', destination_encryption=SSE_B2_AES) self.bucket.copy( b_id, 'e', destination_encryption=SSE_C_AES, file_info={}, content_type='text/plain' ) actual = [info.server_side_encryption for info in self.bucket.list_file_versions('a')][0] - self.assertEqual(SSE_NONE, actual) # bucket default + self.assertEqual(SSE_B2_AES, actual) # bucket default actual = self.bucket.get_file_info_by_name('a').server_side_encryption - self.assertEqual(SSE_NONE, actual) # bucket default + self.assertEqual(SSE_B2_AES, actual) # bucket default actual = [info.server_side_encryption for info in self.bucket.list_file_versions('b')][0] self.assertEqual(SSE_B2_AES, actual) # explicitly requested sse-b2 actual = self.bucket.get_file_info_by_name('b').server_side_encryption self.assertEqual(SSE_B2_AES, actual) # explicitly requested sse-b2 - # actual = [info.server_side_encryption for info in self.bucket.list_file_versions('c')][0] - # self.assertEqual(SSE_NONE, actual) # explicitly requested none - actual = [info.server_side_encryption for info in self.bucket.list_file_versions('d')][0] self.assertEqual(SSE_B2_AES, actual) # explicitly requested sse-b2 actual = self.bucket.get_file_info_by_name('d').server_side_encryption @@ -1432,7 +1463,7 @@ def test_copy_encryption(self): data = b'hello_world' a = self.bucket.upload_bytes(data, 'a') a_id = a.id_ - self.assertEqual(a.server_side_encryption, SSE_NONE) + self.assertEqual(a.server_side_encryption, SSE_B2_AES) b = self.bucket.upload_bytes(data, 'b', encryption=SSE_B2_AES) self.assertEqual(b.server_side_encryption, SSE_B2_AES) @@ -1463,8 +1494,8 @@ def test_copy_encryption(self): ), SSE_C_AES_NO_SECRET, ), - (dict(file_id=b_id), SSE_NONE), - (dict(file_id=b_id, source_encryption=SSE_B2_AES), SSE_NONE), + (dict(file_id=b_id), SSE_B2_AES), + (dict(file_id=b_id, source_encryption=SSE_B2_AES), SSE_B2_AES), ( dict( file_id=b_id, @@ -1500,7 +1531,7 @@ def test_copy_encryption(self): file_info={'new': 'value'}, content_type='text/plain', ), - SSE_NONE, + SSE_B2_AES, ), ( dict( @@ -1509,7 +1540,7 @@ def test_copy_encryption(self): source_file_info={'old': 'value'}, source_content_type='text/plain', ), - SSE_NONE, + SSE_B2_AES, ), ( dict( @@ -1569,6 +1600,147 @@ def _make_file(self, bucket=None): return actual_bucket.upload_bytes(data, 'hello.txt').id_ +class FileCreationEncryptionTests: + EXPECTED_DEFAULT_ENCRYPTION = None + + def _file_creators(self): + data = b'hello world' + part_size = self.simulator.MIN_PART_SIZE + large_data = self._make_data(part_size * 3) + + def upload_local_file(file_name, **kwargs): + with tempfile.NamedTemporaryFile() as local_file: + local_file.write(data) + local_file.flush() + return self.bucket.upload_local_file(local_file.name, file_name, **kwargs) + + def copy(file_name, source_data=data, **kwargs): + source = self.bucket.upload_bytes(source_data, f'{file_name}-source') + if 'encryption' in kwargs: + kwargs['destination_encryption'] = kwargs.pop('encryption') + return self.bucket.copy(source.id_, file_name, **kwargs) + + return { + 'upload_bytes': lambda file_name, **kwargs: self.bucket.upload_bytes( + data, file_name, **kwargs + ), + 'upload_local_file': upload_local_file, + 'upload_unbound_stream': lambda file_name, **kwargs: self.bucket.upload_unbound_stream( + io.BytesIO(data), file_name, **kwargs + ), + 'upload': lambda file_name, **kwargs: self.bucket.upload( + UploadSourceBytes(data), file_name, **kwargs + ), + 'create_file': lambda file_name, **kwargs: self.bucket.create_file( + [WriteIntent(UploadSourceBytes(data))], file_name, **kwargs + ), + 'create_file_stream': lambda file_name, **kwargs: self.bucket.create_file_stream( + iter([WriteIntent(UploadSourceBytes(data))]), file_name, **kwargs + ), + 'concatenate': lambda file_name, **kwargs: self.bucket.concatenate( + [UploadSourceBytes(data)], file_name, **kwargs + ), + 'concatenate_stream': lambda file_name, **kwargs: self.bucket.concatenate_stream( + iter([UploadSourceBytes(data)]), file_name, **kwargs + ), + 'copy': copy, + 'large_upload': lambda file_name, **kwargs: self.bucket.upload_unbound_stream( + io.BytesIO(large_data), + file_name, + recommended_upload_part_size=part_size, + **kwargs, + ), + 'large_copy': lambda file_name, **kwargs: copy( + file_name, + source_data=large_data, + length=len(large_data), + min_part_size=part_size, + max_part_size=part_size, + **kwargs, + ), + } + + def test_file_creation_uses_bucket_default_encryption(self): + for method_name, create_file in self._file_creators().items(): + with self.subTest(method=method_name): + file_version = create_file(f'{method_name}-default') + self.assertEqual( + file_version.server_side_encryption, self.EXPECTED_DEFAULT_ENCRYPTION + ) + + def test_file_creation_accepts_explicit_sse_b2(self): + for method_name, create_file in self._file_creators().items(): + with self.subTest(method=method_name): + file_version = create_file(f'{method_name}-sse-b2', encryption=SSE_B2_AES) + self.assertEqual(file_version.server_side_encryption, SSE_B2_AES) + + def test_file_creation_with_explicit_no_encryption(self): + for method_name, create_file in self._file_creators().items(): + with self.subTest(method=method_name): + with self.assertRaises(WrongEncryptionSettingForFileWrite): + create_file(f'{method_name}-none', encryption=SSE_NONE) + + RAW_FILE_WRITE_METHODS = ( + 'upload_file', + 'upload_part', + 'start_large_file', + 'copy_file', + 'copy_part', + ) + RAW_ENCRYPTION_ARGUMENTS = ('server_side_encryption', 'destination_server_side_encryption') + + def test_file_creation_without_encryption_passes_none_to_raw_api(self): + """ + An omitted encryption setting must reach the raw API as None, so the server + applies the bucket default; nothing along the way may fill in a value. + """ + for method_name, create_file in self._file_creators().items(): + with self.subTest(method=method_name): + with self._record_raw_api_calls(self.RAW_FILE_WRITE_METHODS) as calls: + create_file(f'{method_name}-omitted') + self.assertTrue(calls, 'no raw write reached the simulator') + for raw_method_name, arguments in calls: + for argument in self.RAW_ENCRYPTION_ARGUMENTS: + if argument in arguments: + self.assertIsNone( + arguments[argument], f'{raw_method_name} received {argument}' + ) + + +class TestFileCreationEncryption(FileCreationEncryptionTests, TestCaseWithBucket): + EXPECTED_DEFAULT_ENCRYPTION = SSE_B2_AES + + +class TestBucketEncryptionOmitted(TestCaseWithBucket): + """ + An omitted default encryption must reach the raw API as None, so the server + applies or keeps its own default; nothing along the way may fill in a value. + """ + + def test_create_bucket_without_encryption_passes_none_to_raw_api(self): + with self._record_raw_api_calls(('create_bucket',)) as calls: + self.api.create_bucket('another-bucket', 'allPrivate') + + self.assertEqual( + [arguments['default_server_side_encryption'] for _, arguments in calls], [None] + ) + + def test_update_bucket_without_encryption_passes_none_to_raw_api(self): + updates = { + 'update': lambda: self.bucket.update(bucket_info={'updated': 'true'}), + 'set_info': lambda: self.bucket.set_info({'updated': 'true'}), + 'set_type': lambda: self.bucket.set_type('allPrivate'), + } + for method_name, update in updates.items(): + with self.subTest(method=method_name): + with self._record_raw_api_calls(('update_bucket',)) as calls: + update() + self.assertEqual( + [arguments['default_server_side_encryption'] for _, arguments in calls], + [None], + ) + + class TestUpdate(TestCaseWithBucket): def test_update(self): result = self.bucket.update( @@ -1723,7 +1895,6 @@ def test_upload_bytes(self): self.assertTrue(isinstance(file_info, VFileVersionInfo)) self._check_file_contents('file1', data) self._check_large_file_sha1('file1', None) - self.assertEqual(file_info.server_side_encryption, SSE_NONE) def test_upload_bytes_file_retention(self): data = b'hello world' @@ -1819,9 +1990,12 @@ def test_upload_local_file(self): self._check_file_contents('file1', data) self._check_large_file_sha1('file1', None) self.assertTrue(isinstance(file_info, VFileVersionInfo)) - self.assertEqual(file_info.server_side_encryption, SSE_NONE) + self.assertEqual(file_info.server_side_encryption, SSE_B2_AES) print(file_info.as_dict()) - self.assertEqual(file_info.as_dict()['serverSideEncryption'], {'mode': 'none'}) + self.assertEqual( + file_info.as_dict()['serverSideEncryption'], + {'algorithm': 'AES256', 'mode': 'SSE-B2'}, + ) @pytest.mark.apiver(from_ver=2) def test_upload_local_file_incremental(self): @@ -1991,7 +2165,7 @@ def test_create_file_over_10k_parts(self): created_file.size, created_file.server_side_encryption, ) - expected = ('9998', 'created_file', len(data), SSE_NONE) + expected = ('9998', 'created_file', len(data), SSE_B2_AES) self.assertEqual(expected, actual) def test_upload_large_resume(self): @@ -2230,7 +2404,7 @@ def test_create_remote(self): created_file.size, created_file.server_side_encryption, ) - expected = ('9997', 'created_file', 33, SSE_NONE) + expected = ('9997', 'created_file', 33, SSE_B2_AES) self.assertEqual(expected, actual) def test_create_remote_encryption(self): diff --git a/test/unit/replication/test_monitoring.py b/test/unit/replication/test_monitoring.py index a035e68aa..f4955dae2 100644 --- a/test/unit/replication/test_monitoring.py +++ b/test/unit/replication/test_monitoring.py @@ -31,7 +31,7 @@ DEFAULT_REPLICATION_RESULT = dict( source_replication_status=None, source_has_hide_marker=False, - source_encryption_mode=EncryptionMode.NONE, + source_encryption_mode=EncryptionMode.SSE_B2, source_has_large_metadata=False, source_has_file_retention=False, source_has_legal_hold=False, @@ -102,19 +102,7 @@ def test_scan_source(source_bucket, test_file, monitor): ] report = monitor.scan(scan_destination=False) - assert report.counter_by_status[ReplicationScanResult(**DEFAULT_REPLICATION_RESULT)] == 2 - - assert ( - report.counter_by_status[ - ReplicationScanResult( - **{ - **DEFAULT_REPLICATION_RESULT, - 'source_encryption_mode': EncryptionMode.SSE_B2, - } - ) - ] - == 1 - ) + assert report.counter_by_status[ReplicationScanResult(**DEFAULT_REPLICATION_RESULT)] == 3 assert ( report.counter_by_status[ @@ -183,7 +171,7 @@ def test_scan_source(source_bucket, test_file, monitor): **DEFAULT_REPLICATION_RESULT, ) ][0] - == files[1] + == files[2] ) diff --git a/test/unit/test_raw_api.py b/test/unit/test_raw_api.py new file mode 100644 index 000000000..4745f02ed --- /dev/null +++ b/test/unit/test_raw_api.py @@ -0,0 +1,171 @@ +###################################################################### +# +# File: test/unit/test_raw_api.py +# +# Copyright 2026 Backblaze Inc. All Rights Reserved. +# +# License https://www.backblaze.com/using_b2_code.html +# +###################################################################### +from __future__ import annotations + +import io + +import pytest +from apiver_deps import ( + B2RawHTTPApi, + EncryptionMode, + EncryptionSetting, +) +from apiver_deps_exception import WrongEncryptionSettingForFileWrite + +NO_ENCRYPTION = EncryptionSetting(EncryptionMode.NONE) + + +@pytest.fixture +def raw_api(mocker): + return B2RawHTTPApi(mocker.MagicMock()) + + +@pytest.mark.parametrize( + 'write_file', + [ + pytest.param( + lambda raw_api: raw_api.upload_file( + 'upload-url', + 'upload-token', + 'file-name', + 1, + 'text/plain', + 'sha1', + {}, + io.BytesIO(), + server_side_encryption=NO_ENCRYPTION, + ), + id='upload_file', + ), + pytest.param( + lambda raw_api: raw_api.upload_part( + 'upload-url', + 'upload-token', + 1, + 1, + 'sha1', + io.BytesIO(), + server_side_encryption=NO_ENCRYPTION, + ), + id='upload_part', + ), + pytest.param( + lambda raw_api: raw_api.start_large_file( + 'api-url', + 'account-token', + 'bucket-id', + 'file-name', + 'text/plain', + {}, + server_side_encryption=NO_ENCRYPTION, + ), + id='start_large_file', + ), + pytest.param( + lambda raw_api: raw_api.copy_file( + 'api-url', + 'account-token', + 'source-file-id', + 'new-file-name', + destination_server_side_encryption=NO_ENCRYPTION, + ), + id='copy_file', + ), + pytest.param( + lambda raw_api: raw_api.copy_part( + 'api-url', + 'account-token', + 'source-file-id', + 'large-file-id', + 1, + destination_server_side_encryption=NO_ENCRYPTION, + ), + id='copy_part', + ), + ], +) +def test_raw_api_rejects_no_encryption_for_file_writes(raw_api, write_file): + with pytest.raises(WrongEncryptionSettingForFileWrite): + write_file(raw_api) + + +@pytest.mark.parametrize( + 'call, transport_method', + [ + pytest.param( + lambda raw_api: raw_api.create_bucket( + 'api-url', 'account-token', 'account-id', 'bucket-name', 'allPrivate' + ), + 'post_json_return_json', + id='create_bucket', + ), + pytest.param( + lambda raw_api: raw_api.update_bucket( + 'api-url', 'account-token', 'account-id', 'bucket-id', bucket_type='allPrivate' + ), + 'post_json_return_json', + id='update_bucket', + ), + pytest.param( + lambda raw_api: raw_api.upload_file( + 'upload-url', + 'upload-token', + 'file-name', + 1, + 'text/plain', + 'sha1', + {}, + io.BytesIO(), + ), + 'post_content_return_json', + id='upload_file', + ), + pytest.param( + lambda raw_api: raw_api.upload_part( + 'upload-url', 'upload-token', 1, 1, 'sha1', io.BytesIO() + ), + 'post_content_return_json', + id='upload_part', + ), + pytest.param( + lambda raw_api: raw_api.start_large_file( + 'api-url', 'account-token', 'bucket-id', 'file-name', 'text/plain', {} + ), + 'post_json_return_json', + id='start_large_file', + ), + pytest.param( + lambda raw_api: raw_api.copy_file( + 'api-url', 'account-token', 'source-file-id', 'new-file-name' + ), + 'post_json_return_json', + id='copy_file', + ), + pytest.param( + lambda raw_api: raw_api.copy_part( + 'api-url', 'account-token', 'source-file-id', 'large-file-id', 1 + ), + 'post_json_return_json', + id='copy_part', + ), + ], +) +def test_raw_api_omits_encryption_from_request_when_not_given(raw_api, call, transport_method): + """ + With no encryption setting given, nothing about encryption may appear in the + request: the server applies the bucket default. + """ + call(raw_api) + + transport_call = getattr(raw_api.b2_http, transport_method).call_args + _, headers, payload = transport_call.args[:3] + sent = {**headers, **(payload if isinstance(payload, dict) else {})} + + assert not [key for key in sent if 'encryption' in key.lower()], sent diff --git a/test/unit/test_raw_simulator.py b/test/unit/test_raw_simulator.py index 64350c9a9..a14a6f120 100644 --- a/test/unit/test_raw_simulator.py +++ b/test/unit/test_raw_simulator.py @@ -10,13 +10,18 @@ from __future__ import annotations import hashlib +import io from unittest.mock import Mock import pytest +from apiver_deps import EncryptionMode, EncryptionSetting +from apiver_deps_exception import WrongEncryptionSettingForFileWrite from b2sdk import v3 from test.helpers import patch_bind_params +NO_ENCRYPTION = EncryptionSetting(EncryptionMode.NONE) + @pytest.fixture def dummy_bucket_simulator(apiver_module): @@ -127,3 +132,72 @@ def test_raw_simulator__upload_file__supports_file_infos(dummy_raw_simulator, fi ) assert mock_method.get_bound_call_args()['file_info'] == file_info assert 'file_infos' not in mock_method.call_args[1] + + +@pytest.mark.parametrize( + 'write_file', + [ + pytest.param( + lambda raw_simulator: raw_simulator.upload_file( + 'upload-url', + 'upload-token', + 'file-name', + 1, + 'text/plain', + 'sha1', + {}, + io.BytesIO(), + server_side_encryption=NO_ENCRYPTION, + ), + id='upload_file', + ), + pytest.param( + lambda raw_simulator: raw_simulator.upload_part( + 'upload-url', + 'upload-token', + 1, + 1, + 'sha1', + io.BytesIO(), + server_side_encryption=NO_ENCRYPTION, + ), + id='upload_part', + ), + pytest.param( + lambda raw_simulator: raw_simulator.start_large_file( + 'api-url', + 'account-token', + 'bucket-id', + 'file-name', + 'text/plain', + {}, + server_side_encryption=NO_ENCRYPTION, + ), + id='start_large_file', + ), + pytest.param( + lambda raw_simulator: raw_simulator.copy_file( + 'api-url', + 'account-token', + 'source-file-id', + 'new-file-name', + destination_server_side_encryption=NO_ENCRYPTION, + ), + id='copy_file', + ), + pytest.param( + lambda raw_simulator: raw_simulator.copy_part( + 'api-url', + 'account-token', + 'source-file-id', + 'large-file-id', + 1, + destination_server_side_encryption=NO_ENCRYPTION, + ), + id='copy_part', + ), + ], +) +def test_raw_simulator_rejects_no_encryption_for_file_writes(dummy_raw_simulator, write_file): + with pytest.raises(WrongEncryptionSettingForFileWrite): + write_file(dummy_raw_simulator) diff --git a/test/unit/v0/test_bucket.py b/test/unit/v0/test_bucket.py index 4d62dcb3a..6c1aa2c6d 100644 --- a/test/unit/v0/test_bucket.py +++ b/test/unit/v0/test_bucket.py @@ -21,7 +21,6 @@ from .deps import ( NO_RETENTION_FILE_SETTING, SSE_B2_AES, - SSE_NONE, AbstractProgressListener, B2Api, BucketSimulator, @@ -61,6 +60,7 @@ MaxRetriesExceeded, SSECKeyError, UnsatisfiableRange, + WrongEncryptionSettingForFileWrite, ) SSE_C_AES = EncryptionSetting( @@ -305,7 +305,7 @@ def test_version_by_name(self): 11, 'upload', 'b2/x-auto', - 'none', + 'SSE-B2', NO_RETENTION_FILE_SETTING, LegalHold.UNSET, None, @@ -465,11 +465,10 @@ def test_encryption(self): data = b'hello world' a = self.bucket.upload_bytes(data, 'a') a_id = a.id_ - self.assertEqual(a.server_side_encryption, SSE_NONE) + self.assertEqual(a.server_side_encryption, SSE_B2_AES) b = self.bucket.upload_bytes(data, 'b', encryption=SSE_B2_AES) self.assertEqual(b.server_side_encryption, SSE_B2_AES) b_id = b.id_ - # c_id = self.bucket.upload_bytes(data, 'c', encryption=SSE_NONE).id_ # TODO self.bucket.copy(a_id, 'd', destination_encryption=SSE_B2_AES) self.bucket.copy( b_id, 'e', destination_encryption=SSE_C_AES, file_info={}, content_type='text/plain' @@ -477,16 +476,12 @@ def test_encryption(self): actual = [info for info in self.bucket.list_file_versions('a')['files']][0] actual = EncryptionSettingFactory.from_file_version_dict(actual) - self.assertEqual(SSE_NONE, actual) # bucket default + self.assertEqual(SSE_B2_AES, actual) # bucket default actual = [info for info in self.bucket.list_file_versions('b')['files']][0] actual = EncryptionSettingFactory.from_file_version_dict(actual) self.assertEqual(SSE_B2_AES, actual) # explicitly requested sse-b2 - # actual = [info for info in self.bucket.list_file_versions('c')][0] - # actual = EncryptionSettingFactory.from_file_version_dict(actual) - # self.assertEqual(SSE_NONE, actual) # explicitly requested none - actual = [info for info in self.bucket.list_file_versions('d')['files']][0] actual = EncryptionSettingFactory.from_file_version_dict(actual) self.assertEqual(SSE_B2_AES, actual) # explicitly requested sse-b2 @@ -617,7 +612,7 @@ def test_copy_encryption(self): data = b'hello_world' a = self.bucket.upload_bytes(data, 'a') a_id = a.id_ - self.assertEqual(a.server_side_encryption, SSE_NONE) + self.assertEqual(a.server_side_encryption, SSE_B2_AES) b = self.bucket.upload_bytes(data, 'b', encryption=SSE_B2_AES) self.assertEqual(b.server_side_encryption, SSE_B2_AES) @@ -648,8 +643,8 @@ def test_copy_encryption(self): ), SSE_C_AES_NO_SECRET, ), - (dict(file_id=b_id), SSE_NONE), - (dict(file_id=b_id, source_encryption=SSE_B2_AES), SSE_NONE), + (dict(file_id=b_id), SSE_B2_AES), + (dict(file_id=b_id, source_encryption=SSE_B2_AES), SSE_B2_AES), ( dict( file_id=b_id, @@ -685,7 +680,7 @@ def test_copy_encryption(self): file_info={'new': 'value'}, content_type='text/plain', ), - SSE_NONE, + SSE_B2_AES, ), ( dict( @@ -694,7 +689,7 @@ def test_copy_encryption(self): source_file_info={'old': 'value'}, source_content_type='text/plain', ), - SSE_NONE, + SSE_B2_AES, ), ( dict( @@ -761,6 +756,19 @@ def test_upload_bytes(self): self.assertTrue(isinstance(file_info, FileVersionInfo)) self._check_file_contents('file1', data) + def test_upload_bytes_defaults_to_sse_b2(self): + file_info = self.bucket.upload_bytes(b'hello world', 'file1') + + self.assertEqual(file_info.server_side_encryption, SSE_B2_AES) + + def test_upload_bytes_rejects_plaintext_encryption(self): + with pytest.raises(WrongEncryptionSettingForFileWrite): + self.bucket.upload_bytes( + b'hello world', + 'file1', + encryption=EncryptionSetting(mode=EncryptionMode.NONE), + ) + def test_upload_bytes_file_retention(self): data = b'hello world' retention = FileRetentionSetting(RetentionMode.COMPLIANCE, 150) @@ -803,6 +811,18 @@ def test_upload_local_file_sse_c(self): self.assertEqual(SSE_C_AES_NO_SECRET, file_info.server_side_encryption) self._check_file_contents('file1', data) + def test_upload_local_file_rejects_plaintext_encryption(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, 'file1') + write_file(path, b'hello world') + + with pytest.raises(WrongEncryptionSettingForFileWrite): + self.bucket.upload_local_file( + path, + 'file1', + encryption=EncryptionSetting(mode=EncryptionMode.NONE), + ) + def test_upload_local_file_retention(self): with tempfile.TemporaryDirectory() as d: path = os.path.join(d, 'file1') @@ -846,8 +866,9 @@ def test_upload_local_file(self): path = os.path.join(d, 'file1') data = b'hello world' write_file(path, data) - self.bucket.upload_local_file(path, 'file1') + file_info = self.bucket.upload_local_file(path, 'file1') self._check_file_contents('file1', data) + self.assertEqual(file_info.server_side_encryption, SSE_B2_AES) @pytest.mark.skipif(platform.system() == 'Windows', reason='no os.mkfifo() on Windows') def test_upload_fifo(self): @@ -1027,7 +1048,7 @@ def test_create_remote(self): created_file.size, created_file.server_side_encryption, ) - expected = ('9997', 'created_file', 33, SSE_NONE) + expected = ('9997', 'created_file', 33, SSE_B2_AES) self.assertEqual(expected, actual) def test_create_remote_encryption(self): diff --git a/test/unit/v0/test_raw_api.py b/test/unit/v0/test_raw_api.py index e41f8d95d..e96b527e3 100644 --- a/test/unit/v0/test_raw_api.py +++ b/test/unit/v0/test_raw_api.py @@ -153,6 +153,9 @@ def test_assertion_not_raises(self, bucket_type, bucket_info, default_retention) EncryptionSetting( mode=EncryptionMode.UNKNOWN, ), + EncryptionSetting( + mode=EncryptionMode.NONE, + ), ), ) def test_update_bucket_wrong_encryption(self, encryption_setting): @@ -181,6 +184,9 @@ class TestCreateBucket(BucketTestBase): EncryptionSetting( mode=EncryptionMode.UNKNOWN, ), + EncryptionSetting( + mode=EncryptionMode.NONE, + ), ), ) def test_create_bucket_wrong_encryption(self, encryption_setting): diff --git a/test/unit/v1/test_raw_api.py b/test/unit/v1/test_raw_api.py index af2ce3288..4202fa653 100644 --- a/test/unit/v1/test_raw_api.py +++ b/test/unit/v1/test_raw_api.py @@ -148,6 +148,9 @@ def test_assertion_not_raises(self, bucket_type, bucket_info, default_retention) EncryptionSetting( mode=EncryptionMode.UNKNOWN, ), + EncryptionSetting( + mode=EncryptionMode.NONE, + ), ), ) def test_update_bucket_wrong_encryption(self, encryption_setting): @@ -176,6 +179,9 @@ class TestCreateBucket(BucketTestBase): EncryptionSetting( mode=EncryptionMode.UNKNOWN, ), + EncryptionSetting( + mode=EncryptionMode.NONE, + ), ), ) def test_create_bucket_wrong_encryption(self, encryption_setting): From 4015eb7b6a20377538111279d2980ce8228df3de Mon Sep 17 00:00:00 2001 From: Kordian Kowalski Date: Wed, 9 Sep 2026 17:29:12 +0200 Subject: [PATCH 2/2] fix Windows unit failures: don't reopen a NamedTemporaryFile by name --- test/unit/bucket/test_bucket.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/test/unit/bucket/test_bucket.py b/test/unit/bucket/test_bucket.py index 7633b66d1..ead9a449a 100644 --- a/test/unit/bucket/test_bucket.py +++ b/test/unit/bucket/test_bucket.py @@ -1609,10 +1609,10 @@ def _file_creators(self): large_data = self._make_data(part_size * 3) def upload_local_file(file_name, **kwargs): - with tempfile.NamedTemporaryFile() as local_file: - local_file.write(data) - local_file.flush() - return self.bucket.upload_local_file(local_file.name, file_name, **kwargs) + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, 'file1') + write_file(path, data) + return self.bucket.upload_local_file(path, file_name, **kwargs) def copy(file_name, source_data=data, **kwargs): source = self.bucket.upload_bytes(source_data, f'{file_name}-source')