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