encryption defaults to SSE-B2; "no encryption" is disallowed

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:

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,))

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
 

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,18 +1086,18 @@ 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),
             'X-Bz-Part-Number': str(part_number),
             '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()
             )

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
         ):

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',
 )

changelog.d/+file-write-encryption-check.added.md

@@ -0,0 +1 @@
+Add `EncryptionSetting.can_be_used_for_file_write()` and the `WrongEncryptionSettingForFileWrite` exception.

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.