1# Copyright (c) 2013 Amazon.com, Inc. or its affiliates.  All Rights Reserved
2#
3# Permission is hereby granted, free of charge, to any person obtaining a
4# copy of this software and associated documentation files (the
5# "Software"), to deal in the Software without restriction, including
6# without limitation the rights to use, copy, modify, merge, publish, dis-
7# tribute, sublicense, and/or sell copies of the Software, and to permit
8# persons to whom the Software is furnished to do so, subject to the fol-
9# lowing conditions:
10#
11# The above copyright notice and this permission notice shall be included
12# in all copies or substantial portions of the Software.
13#
14# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
15# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
16# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
17# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
18# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
20# IN THE SOFTWARE.
21#
22import time
23
24from tests.compat import mock, unittest
25from tests.unit import AWSMockServiceTestCase
26from tests.unit import MockServiceWithConfigTestCase
27
28from boto.s3.connection import S3Connection, HostRequiredError
29from boto.s3.connection import S3ResponseError, Bucket
30
31
32class TestSignatureAlteration(AWSMockServiceTestCase):
33    connection_class = S3Connection
34
35    def test_unchanged(self):
36        self.assertEqual(
37            self.service_connection._required_auth_capability(),
38            ['s3']
39        )
40
41    def test_switched(self):
42        conn = self.connection_class(
43            aws_access_key_id='less',
44            aws_secret_access_key='more',
45            host='s3.cn-north-1.amazonaws.com.cn'
46        )
47        self.assertEqual(
48            conn._required_auth_capability(),
49            ['hmac-v4-s3']
50        )
51
52
53class TestSigV4HostError(MockServiceWithConfigTestCase):
54    connection_class = S3Connection
55
56    def test_historical_behavior(self):
57        self.assertEqual(
58            self.service_connection._required_auth_capability(),
59            ['s3']
60        )
61        self.assertEqual(self.service_connection.host, 's3.amazonaws.com')
62
63    def test_sigv4_opt_in(self):
64        # Switch it at the config, so we can check to see how the host is
65        # handled.
66        self.config = {
67            's3': {
68                'use-sigv4': True,
69            }
70        }
71
72        with self.assertRaises(HostRequiredError):
73            # No host+SigV4 == KABOOM
74            self.connection_class(
75                aws_access_key_id='less',
76                aws_secret_access_key='more'
77            )
78
79        # Ensure passing a ``host`` still works.
80        conn = self.connection_class(
81            aws_access_key_id='less',
82            aws_secret_access_key='more',
83            host='s3.cn-north-1.amazonaws.com.cn'
84        )
85        self.assertEqual(
86            conn._required_auth_capability(),
87            ['hmac-v4-s3']
88        )
89        self.assertEqual(
90            conn.host,
91            's3.cn-north-1.amazonaws.com.cn'
92        )
93
94
95class TestSigV4Presigned(MockServiceWithConfigTestCase):
96    connection_class = S3Connection
97
98    def test_sigv4_presign(self):
99        self.config = {
100            's3': {
101                'use-sigv4': True,
102            }
103        }
104
105        conn = self.connection_class(
106            aws_access_key_id='less',
107            aws_secret_access_key='more',
108            host='s3.amazonaws.com'
109        )
110
111        # Here we force an input iso_date to ensure we always get the
112        # same signature.
113        url = conn.generate_url_sigv4(86400, 'GET', bucket='examplebucket',
114            key='test.txt', iso_date='20140625T000000Z')
115
116        self.assertIn('a937f5fbc125d98ac8f04c49e0204ea1526a7b8ca058000a54c192457be05b7d', url)
117
118    def test_sigv4_presign_optional_params(self):
119        self.config = {
120            's3': {
121                'use-sigv4': True,
122            }
123        }
124
125        conn = self.connection_class(
126            aws_access_key_id='less',
127            aws_secret_access_key='more',
128            security_token='token',
129            host='s3.amazonaws.com'
130        )
131
132        url = conn.generate_url_sigv4(86400, 'GET', bucket='examplebucket',
133            key='test.txt', version_id=2)
134
135        self.assertIn('VersionId=2', url)
136        self.assertIn('X-Amz-Security-Token=token', url)
137
138    def test_sigv4_presign_headers(self):
139        self.config = {
140            's3': {
141                'use-sigv4': True,
142            }
143        }
144
145        conn = self.connection_class(
146            aws_access_key_id='less',
147            aws_secret_access_key='more',
148            host='s3.amazonaws.com'
149        )
150
151        headers = {'x-amz-meta-key': 'val'}
152        url = conn.generate_url_sigv4(86400, 'GET', bucket='examplebucket',
153                                      key='test.txt', headers=headers)
154
155        self.assertIn('host', url)
156        self.assertIn('x-amz-meta-key', url)
157
158
159class TestUnicodeCallingFormat(AWSMockServiceTestCase):
160    connection_class = S3Connection
161
162    def default_body(self):
163        return """<?xml version="1.0" encoding="UTF-8"?>
164<ListAllMyBucketsResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
165  <Owner>
166    <ID>bcaf1ffd86f461ca5fb16fd081034f</ID>
167    <DisplayName>webfile</DisplayName>
168  </Owner>
169  <Buckets>
170    <Bucket>
171      <Name>quotes</Name>
172      <CreationDate>2006-02-03T16:45:09.000Z</CreationDate>
173    </Bucket>
174    <Bucket>
175      <Name>samples</Name>
176      <CreationDate>2006-02-03T16:41:58.000Z</CreationDate>
177    </Bucket>
178  </Buckets>
179</ListAllMyBucketsResult>"""
180
181    def create_service_connection(self, **kwargs):
182        kwargs['calling_format'] = u'boto.s3.connection.OrdinaryCallingFormat'
183        return super(TestUnicodeCallingFormat,
184                     self).create_service_connection(**kwargs)
185
186    def test_unicode_calling_format(self):
187        self.set_http_response(status_code=200)
188        self.service_connection.get_all_buckets()
189
190
191class TestHeadBucket(AWSMockServiceTestCase):
192    connection_class = S3Connection
193
194    def default_body(self):
195        # HEAD requests always have an empty body.
196        return ""
197
198    def test_head_bucket_success(self):
199        self.set_http_response(status_code=200)
200        buck = self.service_connection.head_bucket('my-test-bucket')
201        self.assertTrue(isinstance(buck, Bucket))
202        self.assertEqual(buck.name, 'my-test-bucket')
203
204    def test_head_bucket_forbidden(self):
205        self.set_http_response(status_code=403)
206
207        with self.assertRaises(S3ResponseError) as cm:
208            self.service_connection.head_bucket('cant-touch-this')
209
210        err = cm.exception
211        self.assertEqual(err.status, 403)
212        self.assertEqual(err.error_code, 'AccessDenied')
213        self.assertEqual(err.message, 'Access Denied')
214
215    def test_head_bucket_notfound(self):
216        self.set_http_response(status_code=404)
217
218        with self.assertRaises(S3ResponseError) as cm:
219            self.service_connection.head_bucket('totally-doesnt-exist')
220
221        err = cm.exception
222        self.assertEqual(err.status, 404)
223        self.assertEqual(err.error_code, 'NoSuchBucket')
224        self.assertEqual(err.message, 'The specified bucket does not exist')
225
226    def test_head_bucket_other(self):
227        self.set_http_response(status_code=405)
228
229        with self.assertRaises(S3ResponseError) as cm:
230            self.service_connection.head_bucket('you-broke-it')
231
232        err = cm.exception
233        self.assertEqual(err.status, 405)
234        # We don't have special-cases for this error status.
235        self.assertEqual(err.error_code, None)
236        self.assertEqual(err.message, '')
237
238
239if __name__ == "__main__":
240    unittest.main()
241