1###################################################################### 2# 3# File: b2sdk/v1/sync/encryption_provider.py 4# 5# Copyright 2021 Backblaze Inc. All Rights Reserved. 6# 7# License https://www.backblaze.com/using_b2_code.html 8# 9###################################################################### 10 11import inspect 12from abc import abstractmethod 13from typing import Optional 14 15from b2sdk import v2 16from ..bucket import Bucket 17from ..file_version import FileVersionInfo 18 19 20# wrapper to translate new argument names to old ones 21class SyncEncryptionSettingsProviderWrapper(v2.AbstractSyncEncryptionSettingsProvider): 22 def __init__(self, provider): 23 self.provider = provider 24 25 def __repr__(self): 26 return "%s(%s)" % ( 27 self.__class__.__name__, 28 self.provider, 29 ) 30 31 def get_setting_for_upload( 32 self, 33 bucket: Bucket, 34 b2_file_name: str, 35 file_info: Optional[dict], 36 length: int, 37 ) -> Optional[v2.EncryptionSetting]: 38 return self.provider.get_setting_for_upload( 39 bucket=bucket, 40 b2_file_name=b2_file_name, 41 file_info=file_info, 42 length=length, 43 ) 44 45 def get_source_setting_for_copy( 46 self, 47 bucket: Bucket, 48 source_file_version: v2.FileVersion, 49 ) -> Optional[v2.EncryptionSetting]: 50 return self.provider.get_source_setting_for_copy( 51 bucket=bucket, source_file_version_info=source_file_version 52 ) 53 54 def get_destination_setting_for_copy( 55 self, 56 bucket: Bucket, 57 dest_b2_file_name: str, 58 source_file_version: v2.FileVersion, 59 target_file_info: Optional[dict] = None, 60 ) -> Optional[v2.EncryptionSetting]: 61 return self.provider.get_destination_setting_for_copy( 62 bucket=bucket, 63 dest_b2_file_name=dest_b2_file_name, 64 source_file_version_info=source_file_version, 65 target_file_info=target_file_info, 66 ) 67 68 def get_setting_for_download( 69 self, 70 bucket: Bucket, 71 file_version: v2.FileVersion, 72 ) -> Optional[v2.EncryptionSetting]: 73 return self.provider.get_setting_for_download( 74 bucket=bucket, 75 file_version_info=file_version, 76 ) 77 78 79def wrap_if_necessary(provider): 80 if 'file_version' in inspect.getfullargspec(provider.get_setting_for_download).args: 81 return provider 82 return SyncEncryptionSettingsProviderWrapper(provider) 83 84 85# Old signatures 86class AbstractSyncEncryptionSettingsProvider(v2.AbstractSyncEncryptionSettingsProvider): 87 @abstractmethod 88 def get_setting_for_upload( 89 self, 90 bucket: Bucket, 91 b2_file_name: str, 92 file_info: Optional[dict], 93 length: int, 94 ) -> Optional[v2.EncryptionSetting]: 95 """ 96 Return an EncryptionSetting for uploading an object or None if server should decide. 97 """ 98 99 @abstractmethod 100 def get_source_setting_for_copy( 101 self, 102 bucket: Bucket, 103 source_file_version_info: FileVersionInfo, 104 ) -> Optional[v2.EncryptionSetting]: 105 """ 106 Return an EncryptionSetting for a source of copying an object or None if not required 107 """ 108 109 @abstractmethod 110 def get_destination_setting_for_copy( 111 self, 112 bucket: Bucket, 113 dest_b2_file_name: str, 114 source_file_version_info: FileVersionInfo, 115 target_file_info: Optional[dict] = None, 116 ) -> Optional[v2.EncryptionSetting]: 117 """ 118 Return an EncryptionSetting for a destination for copying an object or None if server should decide 119 """ 120 121 @abstractmethod 122 def get_setting_for_download( 123 self, 124 bucket: Bucket, 125 file_version_info: FileVersionInfo, 126 ) -> Optional[v2.EncryptionSetting]: 127 """ 128 Return an EncryptionSetting for downloading an object from, or None if not required 129 """ 130