| Index: third_party/boto/tests/unit/auth/test_sigv4.py
|
| ===================================================================
|
| --- third_party/boto/tests/unit/auth/test_sigv4.py (revision 33376)
|
| +++ third_party/boto/tests/unit/auth/test_sigv4.py (working copy)
|
| @@ -20,12 +20,17 @@
|
| # IN THE SOFTWARE.
|
| #
|
| import copy
|
| +import mock
|
| from mock import Mock
|
| -from tests.unit import unittest
|
| +import os
|
| +from tests.unit import unittest, MockServiceWithConfigTestCase
|
|
|
| from boto.auth import HmacAuthV4Handler
|
| from boto.auth import S3HmacAuthV4Handler
|
| +from boto.auth import detect_potential_s3sigv4
|
| +from boto.auth import detect_potential_sigv4
|
| from boto.connection import HTTPRequest
|
| +from boto.regioninfo import RegionInfo
|
|
|
|
|
| class TestSigV4Handler(unittest.TestCase):
|
| @@ -38,19 +43,29 @@
|
| '/-/vaults/foo/archives', None, {},
|
| {'x-amz-glacier-version': '2012-06-01'}, '')
|
|
|
| + def test_not_adding_empty_qs(self):
|
| + self.provider.security_token = None
|
| + auth = HmacAuthV4Handler('glacier.us-east-1.amazonaws.com', Mock(), self.provider)
|
| + req = copy.copy(self.request)
|
| + auth.add_auth(req)
|
| + self.assertEqual(req.path, '/-/vaults/foo/archives')
|
| +
|
| def test_inner_whitespace_is_collapsed(self):
|
| auth = HmacAuthV4Handler('glacier.us-east-1.amazonaws.com',
|
| Mock(), self.provider)
|
| self.request.headers['x-amz-archive-description'] = 'two spaces'
|
| + self.request.headers['x-amz-quoted-string'] = ' "a b c" '
|
| headers = auth.headers_to_sign(self.request)
|
| self.assertEqual(headers, {'Host': 'glacier.us-east-1.amazonaws.com',
|
| 'x-amz-archive-description': 'two spaces',
|
| - 'x-amz-glacier-version': '2012-06-01'})
|
| + 'x-amz-glacier-version': '2012-06-01',
|
| + 'x-amz-quoted-string': ' "a b c" '})
|
| # Note the single space between the "two spaces".
|
| self.assertEqual(auth.canonical_headers(headers),
|
| 'host:glacier.us-east-1.amazonaws.com\n'
|
| 'x-amz-archive-description:two spaces\n'
|
| - 'x-amz-glacier-version:2012-06-01')
|
| + 'x-amz-glacier-version:2012-06-01\n'
|
| + 'x-amz-quoted-string:"a b c"')
|
|
|
| def test_canonical_query_string(self):
|
| auth = HmacAuthV4Handler('glacier.us-east-1.amazonaws.com',
|
| @@ -64,6 +79,18 @@
|
| query_string = auth.canonical_query_string(request)
|
| self.assertEqual(query_string, 'Foo.1=aaa&Foo.10=zzz')
|
|
|
| + def test_query_string(self):
|
| + auth = HmacAuthV4Handler('sns.us-east-1.amazonaws.com',
|
| + Mock(), self.provider)
|
| + params = {
|
| + 'Message': u'We \u2665 utf-8'.encode('utf-8'),
|
| + }
|
| + request = HTTPRequest(
|
| + 'POST', 'https', 'sns.us-east-1.amazonaws.com', 443,
|
| + '/', None, params, {}, '')
|
| + query_string = auth.query_string(request)
|
| + self.assertEqual(query_string, 'Message=We%20%E2%99%A5%20utf-8')
|
| +
|
| def test_canonical_uri(self):
|
| auth = HmacAuthV4Handler('glacier.us-east-1.amazonaws.com',
|
| Mock(), self.provider)
|
| @@ -431,3 +458,98 @@
|
| request = self.auth.mangle_path_and_params(request)
|
| authed_req = self.auth.canonical_request(request)
|
| self.assertEqual(authed_req, expected)
|
| +
|
| +
|
| +class FakeS3Connection(object):
|
| + def __init__(self, *args, **kwargs):
|
| + self.host = kwargs.pop('host', None)
|
| +
|
| + @detect_potential_s3sigv4
|
| + def _required_auth_capability(self):
|
| + return ['nope']
|
| +
|
| + def _mexe(self, *args, **kwargs):
|
| + pass
|
| +
|
| +
|
| +class FakeEC2Connection(object):
|
| + def __init__(self, *args, **kwargs):
|
| + self.region = kwargs.pop('region', None)
|
| +
|
| + @detect_potential_sigv4
|
| + def _required_auth_capability(self):
|
| + return ['nope']
|
| +
|
| + def _mexe(self, *args, **kwargs):
|
| + pass
|
| +
|
| +
|
| +class TestS3SigV4OptIn(MockServiceWithConfigTestCase):
|
| + connection_class = FakeS3Connection
|
| +
|
| + def test_sigv4_opt_out(self):
|
| + # Default is opt-out.
|
| + fake = FakeS3Connection(host='s3.amazonaws.com')
|
| + self.assertEqual(fake._required_auth_capability(), ['nope'])
|
| +
|
| + def test_sigv4_non_optional(self):
|
| + # Requires SigV4.
|
| + fake = FakeS3Connection(host='s3.cn-north-1.amazonaws.com.cn')
|
| + self.assertEqual(fake._required_auth_capability(), ['hmac-v4-s3'])
|
| +
|
| + def test_sigv4_opt_in_config(self):
|
| + # Opt-in via the config.
|
| + self.config = {
|
| + 's3': {
|
| + 'use-sigv4': True,
|
| + },
|
| + }
|
| + fake = FakeS3Connection()
|
| + self.assertEqual(fake._required_auth_capability(), ['hmac-v4-s3'])
|
| +
|
| + def test_sigv4_opt_in_env(self):
|
| + # Opt-in via the ENV.
|
| + self.environ['S3_USE_SIGV4'] = True
|
| + fake = FakeS3Connection(host='s3.amazonaws.com')
|
| + self.assertEqual(fake._required_auth_capability(), ['hmac-v4-s3'])
|
| +
|
| +
|
| +class TestSigV4OptIn(MockServiceWithConfigTestCase):
|
| + connection_class = FakeEC2Connection
|
| +
|
| + def setUp(self):
|
| + super(TestSigV4OptIn, self).setUp()
|
| + self.standard_region = RegionInfo(
|
| + name='us-west-2',
|
| + endpoint='ec2.us-west-2.amazonaws.com'
|
| + )
|
| + self.sigv4_region = RegionInfo(
|
| + name='cn-north-1',
|
| + endpoint='ec2.cn-north-1.amazonaws.com.cn'
|
| + )
|
| +
|
| + def test_sigv4_opt_out(self):
|
| + # Default is opt-out.
|
| + fake = FakeEC2Connection(region=self.standard_region)
|
| + self.assertEqual(fake._required_auth_capability(), ['nope'])
|
| +
|
| + def test_sigv4_non_optional(self):
|
| + # Requires SigV4.
|
| + fake = FakeEC2Connection(region=self.sigv4_region)
|
| + self.assertEqual(fake._required_auth_capability(), ['hmac-v4'])
|
| +
|
| + def test_sigv4_opt_in_config(self):
|
| + # Opt-in via the config.
|
| + self.config = {
|
| + 'ec2': {
|
| + 'use-sigv4': True,
|
| + },
|
| + }
|
| + fake = FakeEC2Connection(region=self.standard_region)
|
| + self.assertEqual(fake._required_auth_capability(), ['hmac-v4'])
|
| +
|
| + def test_sigv4_opt_in_env(self):
|
| + # Opt-in via the ENV.
|
| + self.environ['EC2_USE_SIGV4'] = True
|
| + fake = FakeEC2Connection(region=self.standard_region)
|
| + self.assertEqual(fake._required_auth_capability(), ['hmac-v4'])
|
|
|