1######################################################################
2#
3# File: test/unit/v1/test_copy_manager.py
4#
5# Copyright 2021 Backblaze Inc. All Rights Reserved.
6#
7# License https://www.backblaze.com/using_b2_code.html
8#
9######################################################################
10
11from ..test_base import TestBase
12
13from .deps_exception import InvalidAuthToken, Unauthorized, SSECKeyIdMismatchInCopy
14from .deps import MetadataDirectiveMode
15from .deps import EncryptionAlgorithm, EncryptionSetting, EncryptionMode, EncryptionKey, SSE_NONE, SSE_B2_AES
16from b2sdk.transfer.outbound.copy_manager import CopyManager
17from b2sdk.http_constants import SSE_C_KEY_ID_FILE_INFO_KEY_NAME
18
19SSE_C_AES = EncryptionSetting(
20    mode=EncryptionMode.SSE_C,
21    algorithm=EncryptionAlgorithm.AES256,
22    key=EncryptionKey(secret=b'some_key', key_id='some-id'),
23)
24SSE_C_AES_2 = EncryptionSetting(
25    mode=EncryptionMode.SSE_C,
26    algorithm=EncryptionAlgorithm.AES256,
27    key=EncryptionKey(secret=b'some_other_key', key_id='some-id-2'),
28)
29
30
31class TestCopyManager(TestBase):
32    def test_establish_sse_c_replace(self):
33        file_info = {'some_key': 'some_value'}
34        content_type = 'text/plain'
35        metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata(
36            MetadataDirectiveMode.REPLACE,
37            destination_file_info=file_info,
38            destination_content_type=content_type,
39            destination_server_side_encryption=SSE_C_AES,
40            source_server_side_encryption=SSE_C_AES_2,
41            source_file_info=None,
42            source_content_type=None,
43        )
44        self.assertEqual(
45            (
46                MetadataDirectiveMode.REPLACE, {
47                    'some_key': 'some_value',
48                    SSE_C_KEY_ID_FILE_INFO_KEY_NAME: 'some-id'
49                }, content_type
50            ), (metadata_directive, new_file_info, new_content_type)
51        )
52
53    def test_establish_sse_c_copy_no_enc(self):
54        file_info = {}
55        content_type = 'text/plain'
56        metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata(
57            MetadataDirectiveMode.COPY,
58            destination_file_info=file_info,
59            destination_content_type=content_type,
60            destination_server_side_encryption=None,
61            source_server_side_encryption=None,
62            source_file_info=None,
63            source_content_type=None,
64        )
65        self.assertEqual(
66            (MetadataDirectiveMode.COPY, {}, content_type),
67            (metadata_directive, new_file_info, new_content_type)
68        )
69
70    def test_establish_sse_c_copy_b2(self):
71        file_info = {}
72        content_type = 'text/plain'
73        metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata(
74            MetadataDirectiveMode.COPY,
75            destination_file_info=file_info,
76            destination_content_type=content_type,
77            destination_server_side_encryption=SSE_B2_AES,
78            source_server_side_encryption=None,
79            source_file_info=None,
80            source_content_type=None,
81        )
82        self.assertEqual(
83            (MetadataDirectiveMode.COPY, {}, content_type),
84            (metadata_directive, new_file_info, new_content_type)
85        )
86
87    def test_establish_sse_c_copy_same_key_id(self):
88        file_info = None
89        content_type = 'text/plain'
90        metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata(
91            MetadataDirectiveMode.COPY,
92            destination_file_info=file_info,
93            destination_content_type=content_type,
94            destination_server_side_encryption=SSE_C_AES,
95            source_server_side_encryption=SSE_C_AES,
96            source_file_info=None,
97            source_content_type=None,
98        )
99        self.assertEqual(
100            (MetadataDirectiveMode.COPY, None, content_type),
101            (metadata_directive, new_file_info, new_content_type)
102        )
103
104    def test_establish_sse_c_copy_sources_given(self):
105        metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata(
106            MetadataDirectiveMode.COPY,
107            destination_file_info=None,
108            destination_content_type=None,
109            destination_server_side_encryption=SSE_C_AES,
110            source_server_side_encryption=SSE_C_AES_2,
111            source_file_info={
112                'some_key': 'some_value',
113                SSE_C_KEY_ID_FILE_INFO_KEY_NAME: 'some-id-2'
114            },
115            source_content_type='text/plain',
116        )
117        self.assertEqual(
118            (
119                MetadataDirectiveMode.REPLACE, {
120                    'some_key': 'some_value',
121                    SSE_C_KEY_ID_FILE_INFO_KEY_NAME: 'some-id'
122                }, 'text/plain'
123            ), (metadata_directive, new_file_info, new_content_type)
124        )
125
126    def test_establish_sse_c_copy_sources_unknown(self):
127        for source_file_info, source_content_type in [
128            (None, None),
129            ({
130                'a': 'b'
131            }, None),
132            (None, 'text/plain'),
133        ]:
134            with self.subTest(
135                source_file_info=source_file_info, source_content_type=source_content_type
136            ):
137                with self.assertRaises(
138                    SSECKeyIdMismatchInCopy,
139                    'attempting to copy file using MetadataDirectiveMode.COPY without providing source_file_info '
140                    'and source_content_type for differing sse_c_key_ids: source="some-id-2", destination="some-id"'
141                ):
142                    CopyManager.establish_sse_c_file_metadata(
143                        MetadataDirectiveMode.COPY,
144                        destination_file_info=None,
145                        destination_content_type=None,
146                        destination_server_side_encryption=SSE_C_AES,
147                        source_server_side_encryption=SSE_C_AES_2,
148                        source_file_info=source_file_info,
149                        source_content_type=source_content_type,
150                    )
151