Allow anonymous calls on an operation-by-operation basis. Fixes boto#206

.

botocore/auth.py

@@ -60,8 +60,6 @@ class SigV2Auth(BaseSigner):
 
     def __init__(self, credentials):
         self.credentials = credentials
-        if self.credentials is None:
-            raise NoCredentialsError
 
     def calc_signature(self, request, params):
         logger.debug("Calculating signature using v2 auth.")
@@ -92,6 +90,8 @@ def add_auth(self, request):
         # Because of this we have to parse the query params
         # from the request body so we can update them with
         # the sigv2 auth params.
+        if self.credentials is None:
+            raise NoCredentialsError
         if request.data:
             # POST
             params = request.data
@@ -112,10 +112,10 @@ def add_auth(self, request):
 class SigV3Auth(BaseSigner):
     def __init__(self, credentials):
         self.credentials = credentials
-        if self.credentials is None:
-            raise NoCredentialsError
 
     def add_auth(self, request):
+        if self.credentials is None:
+            raise NoCredentialsError
         if 'Date' not in request.headers:
             request.headers['Date'] = formatdate(usegmt=True)
         if self.credentials.token:
@@ -138,8 +138,6 @@ class SigV4Auth(BaseSigner):
 
     def __init__(self, credentials, service_name, region_name):
         self.credentials = credentials
-        if self.credentials is None:
-            raise NoCredentialsError
         # We initialize these value here so the unit tests can have
         # valid values.  But these will get overriden in ``add_auth``
         # later for real requests.
@@ -301,6 +299,8 @@ def signature(self, string_to_sign):
         return self._sign(k_signing, string_to_sign, hex=True)
 
     def add_auth(self, request):
+        if self.credentials is None:
+            raise NoCredentialsError
         # Create a new timestamp for each signing event
         now = datetime.datetime.utcnow()
         self.timestamp = now.strftime('%Y%m%dT%H%M%SZ')
@@ -356,8 +356,6 @@ class HmacV1Auth(BaseSigner):
 
     def __init__(self, credentials, service_name=None, region_name=None):
         self.credentials = credentials
-        if self.credentials is None:
-            raise NoCredentialsError
         self.auth_path = None  # see comment in canonical_resource below
 
     def sign_string(self, string_to_sign):
@@ -450,6 +448,8 @@ def get_signature(self, method, split, headers, expires=None):
         return self.sign_string(string_to_sign)
 
     def add_auth(self, request):
+        if self.credentials is None:
+            raise NoCredentialsError
         logger.debug("Calculating signature using hmacv1 auth.")
         split = urlsplit(request.url)
         logger.debug('HTTP request method: %s', request.method)

botocore/endpoint.py

@@ -33,7 +33,7 @@
 import botocore.response
 import botocore.exceptions
 from botocore.auth import AUTH_TYPE_MAPS
-from botocore.exceptions import UnknownSignatureVersionError
+from botocore.exceptions import UnknownSignatureVersionError, NoCredentialsError
 from botocore.awsrequest import AWSRequest
 from botocore.compat import urljoin, json, quote
 
@@ -72,15 +72,20 @@ def __repr__(self):
     def make_request(self, operation, params):
         logger.debug("Making request for %s (verify_ssl=%s) with params: %s",
                      operation, self.verify, params)
+        no_auth = getattr(operation, 'no_auth', False)
+        # There are two situations where we will skip auth:
+        #   1. If the service as no signature_version attribute
+        #   2. If the operation allows anonymous calls
+        do_auth = self.auth and (not no_auth)
         request = self._create_request_object(operation, params)
-        prepared_request = self.prepare_request(request)
+        prepared_request = self.prepare_request(request, do_auth)
         return self._send_request(prepared_request, operation)
 
     def _create_request_object(self, operation, params):
         raise NotImplementedError('_create_request_object')
 
-    def prepare_request(self, request):
-        if self.auth is not None:
+    def prepare_request(self, request, do_auth=True):
+        if do_auth:
             with self._lock:
                 # Parts of the auth signing code aren't thread safe (things
                 # that manipulate .auth_path), so we're using a lock here to

services/sts.extra.json

@@ -10,5 +10,13 @@
         "https"
       ]
     }
+  },
+  "operations": {
+    "AssumeRoleWithSAML": {
+      "no_auth": true
+    },
+    "AssumeRoleWithWebIdentity": {
+      "no_auth": true
+    }
   }
 }

tests/unit/test_endpoint.py

@@ -154,6 +154,7 @@ def setUp(self):
         self.service.session.emit_first_non_none_response.return_value = None
         self.op = Mock()
         self.op.is_streaming.return_value = False
+        self.op.no_auth = False
         self.auth = Mock()
         self.endpoint = self.ENDPOINT_CLASS(
             self.service, 'us-west-2', 'https://ec2.us-west-2.amazonaws.com/',
@@ -204,6 +205,26 @@ def test_make_request_with_no_auth(self):
         self.assertNotIn('Authorization', prepared_request.headers)
 
 
+class TestQueryEndpointAnonymousOp(TestQueryEndpoint):
+
+    def setUp(self):
+        super(TestQueryEndpointAnonymousOp, self).setUp()
+        self.op.no_auth = True
+
+    def test_make_request(self):
+        self.endpoint.make_request(self.op, {})
+        # Should have authenticated the request
+        self.assertFalse(self.auth.add_auth.called)
+        # http_session should be used to send the request.
+        self.assertTrue(self.http_session.send.called)
+        prepared_request = self.http_session.send.call_args[0][0]
+        self.http_session.send.assert_called_with(
+            prepared_request, verify=True, stream=False,
+            proxies={})
+        self.get_response.assert_called_with(self.service.session,
+            self.op, sentinel.HTTP_RETURN_VALUE)
+
+
 class TestJSONEndpoint(TestEndpointBase):
     ENDPOINT_CLASS = JSONEndpoint
 
@@ -217,6 +238,22 @@ def test_make_request(self):
             proxies={})
 
 
+class TestJSONEndpointAnonymousOp(TestJSONEndpoint):
+
+    def setUp(self):
+        super(TestJSONEndpointAnonymousOp, self).setUp()
+        self.op.no_auth = True
+
+    def test_make_request(self):
+        self.endpoint.make_request(self.op, {})
+        self.assertFalse(self.auth.add_auth.called)
+        self.assertTrue(self.http_session.send.called)
+        prepared_request = self.http_session.send.call_args[0][0]
+        self.http_session.send.assert_called_with(
+            prepared_request, verify=True, stream=False,
+            proxies={})
+
+
 class TestRestEndpoint(TestEndpointBase):
     ENDPOINT_CLASS = RestEndpoint
 
@@ -231,6 +268,23 @@ def test_make_request(self):
             proxies={})
 
 
+class TestRestEndpointAnonymousOp(TestRestEndpoint):
+
+    def setUp(self):
+        super(TestRestEndpointAnonymousOp, self).setUp()
+        self.op.no_auth = True
+
+    def test_make_request(self):
+        self.op.http = {'uri': '/foo', 'method': 'POST'}
+        self.endpoint.make_request(self.op, {
+            'headers': {}, 'uri_params': {}, 'payload': None})
+        self.assertFalse(self.auth.add_auth.called)
+        prepared_request = self.http_session.send.call_args[0][0]
+        self.http_session.send.assert_called_with(
+            prepared_request, verify=True, stream=False,
+            proxies={})
+
+
 class TestRetryInterface(BaseSessionTest):
     def setUp(self):
         super(TestRetryInterface, self).setUp()

tests/unit/test_sts_operations.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python
+# Copyright 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a
+# copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish, dis-
+# tribute, sublicense, and/or sell copies of the Software, and to permit
+# persons to whom the Software is furnished to do so, subject to the fol-
+# lowing conditions:
+#
+# The above copyright notice and this permission notice shall be included
+# in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
+# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+# IN THE SOFTWARE.
+#
+import unittest
+import logging
+
+import mock
+
+from tests import BaseSessionTest
+
+import botocore.session
+from botocore.exceptions import NoCredentialsError
+
+LOG = logging.getLogger(__name__)
+
+
+class TestSTSOperationsWithCreds(BaseSessionTest):
+
+    def setUp(self):
+        super(TestSTSOperationsWithCreds, self).setUp()
+        self.sns = self.session.get_service('sts')
+        self.called_params = {}
+        self.session.register('before-auth.sts',
+                              lambda **kwargs: self.called_params.update(kwargs))
+
+    def reset_called_params(self):
+        self.called_params = {}
+
+    def get_mocked_endpoint(self):
+        endpoint = self.sns.get_endpoint()
+        endpoint._send_request = mock.Mock()
+        return endpoint
+
+    def test_get_session_token(self):
+        op = self.sns.get_operation('GetSessionToken')
+        params = {}
+        endpoint = self.get_mocked_endpoint()
+        endpoint.make_request(op, params)
+        LOG.debug(self.called_params)
+        self.assertIn('auth', self.called_params)
+        self.reset_called_params()
+
+    def test_assume_role_with_saml(self):
+        op = self.sns.get_operation('AssumeRoleWithSAML')
+        self.assertEqual(op.no_auth, True)
+        endpoint = self.get_mocked_endpoint()
+        params = op.build_parameters(principal_arn='principal_arn',
+                                     role_arn='role_arn',
+                                     saml_assertion='saml_assertion')
+        endpoint.make_request(op, params)
+        LOG.debug(self.called_params)
+        self.assertNotIn('auth', self.called_params)
+        self.reset_called_params()
+
+
+class NoCredentialsTest(TestSTSOperationsWithCreds):
+
+    def setUp(self):
+        # Automatically patches out get_credentials to always
+        # return None.
+        super(NoCredentialsTest, self).setUp()
+        self.get_credentials_patch = mock.patch(
+            'botocore.credentials.get_credentials',
+            self.mock_get_credentials)
+
+    def mock_get_credentials(self, session, metadata=None):
+        return None
+
+    def test_get_session_token(self):
+        with self.get_credentials_patch as mock_fn:
+            session = botocore.session.get_session()
+            sns = session.get_service('sts')
+            op = sns.get_operation('GetSessionToken')
+            params = {}
+            endpoint = self.get_mocked_endpoint()
+            self.assertRaises(NoCredentialsError,
+                              endpoint.make_request,
+                              op, params)
+
+    def test_assume_role_with_saml(self):
+        with self.get_credentials_patch as mock_fn:
+            session = botocore.session.get_session()
+            sns = session.get_service('sts')
+            op = sns.get_operation('AssumeRoleWithSAML')
+            self.assertEqual(op.no_auth, True)
+            endpoint = self.get_mocked_endpoint()
+            params = op.build_parameters(principal_arn='principal_arn',
+                                         role_arn='role_arn',
+                                         saml_assertion='saml_assertion')
+            endpoint.make_request(op, params)
+            self.assertNotIn('auth', self.called_params)
+            self.reset_called_params()
+
+
+if __name__ == "__main__":
+    unittest.main()