######################################################################
#
# File: b2sdk/sync/encryption_provider.py
#
# Copyright 2021 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################
from abc import ABCMeta, abstractmethod
from typing import Dict, Optional
from ..encryption.setting import EncryptionSetting
from ..bucket import Bucket
from ..file_version import FileVersionInfo
[docs]class AbstractSyncEncryptionSettingsProvider(metaclass=ABCMeta):
"""
Object which provides an appropriate EncryptionSetting object
for sync, i.e. complex operations with multiple sources and destinations
"""
[docs] @abstractmethod
def get_setting_for_upload(
self,
bucket: Bucket,
b2_file_name: str,
file_info: Optional[dict],
length: int,
) -> Optional[EncryptionSetting]:
"""
Return an EncryptionSetting for uploading an object or None if server should decide.
WARNING: the signature of this method is not final yet and not part of the public interface
"""
[docs] @abstractmethod
def get_source_setting_for_copy(
self,
bucket: Bucket,
source_file_version_info: FileVersionInfo,
) -> Optional[EncryptionSetting]:
"""
Return an EncryptionSetting for a source of copying an object or None if not required
WARNING: the signature of this method is not final yet and not part of the public interface
"""
[docs] @abstractmethod
def get_destination_setting_for_copy(
self,
bucket: Bucket,
dest_b2_file_name: str,
source_file_version_info: FileVersionInfo,
target_file_info: Optional[dict] = None,
) -> Optional[EncryptionSetting]:
"""
Return an EncryptionSetting for a destination for copying an object or None if server should decide
WARNING: the signature of this method is not final yet and not part of the public interface
"""
[docs] @abstractmethod
def get_setting_for_download(
self,
bucket: Bucket,
file_version_info: FileVersionInfo,
) -> Optional[EncryptionSetting]:
"""
Return an EncryptionSetting for downloading an object from, or None if not required
WARNING: the signature of this method is not final yet and not part of the public interface
"""
[docs]class ServerDefaultSyncEncryptionSettingsProvider(AbstractSyncEncryptionSettingsProvider):
"""
Encryption settings provider which assumes setting-less reads
and a bucket default for writes.
"""
[docs] def get_setting_for_upload(self, *args, **kwargs) -> None:
return None
[docs] def get_source_setting_for_copy(self, *args, **kwargs) -> None:
return None
[docs] def get_destination_setting_for_copy(self, *args, **kwargs) -> None:
return None
[docs] def get_setting_for_download(self, *args, **kwargs) -> None:
return None
SERVER_DEFAULT_SYNC_ENCRYPTION_SETTINGS_PROVIDER = ServerDefaultSyncEncryptionSettingsProvider()
[docs]class BasicSyncEncryptionSettingsProvider(AbstractSyncEncryptionSettingsProvider):
"""
Basic encryption setting provider that supports exactly one encryption setting per bucket for reading
and one encryption setting per bucket for writing
WARNING: This class can be used by B2CLI for SSE-B2, but it's still in development
"""
def __init__(
self,
read_bucket_settings: Dict[str, Optional[EncryptionSetting]],
write_bucket_settings: Dict[str, Optional[EncryptionSetting]],
):
self.read_bucket_settings = read_bucket_settings
self.write_bucket_settings = write_bucket_settings
[docs] def get_setting_for_upload(self, bucket, *args, **kwargs) -> Optional[EncryptionSetting]:
return self.write_bucket_settings.get(bucket.name)
[docs] def get_source_setting_for_copy(self, bucket, *args, **kwargs) -> None:
return self.read_bucket_settings.get(bucket.name)
[docs] def get_destination_setting_for_copy(self, bucket, *args,
**kwargs) -> Optional[EncryptionSetting]:
return self.write_bucket_settings.get(bucket.name)
[docs] def get_setting_for_download(self, bucket, *args, **kwargs) -> None:
return self.read_bucket_settings.get(bucket.name)
def __repr__(self):
return '<%s:%s>' % (self.__class__.__name__, self.bucket_settings)