# Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or # implied. # See the License for the specific language governing permissions and # limitations under the License. from unittest import mock import fixtures import uuid import six from six.moves.urllib import parse as urlparse from testtools import matchers from keystoneclient import session from keystoneclient.tests.unit.v3 import client_fixtures from keystoneclient.tests.unit.v3 import utils from keystoneclient import utils as client_utils from keystoneclient.v3.contrib.oauth1 import access_tokens from keystoneclient.v3.contrib.oauth1 import auth from keystoneclient.v3.contrib.oauth1 import consumers from keystoneclient.v3.contrib.oauth1 import request_tokens try: from oauthlib import oauth1 except ImportError: oauth1 = None class ConsumerTests(utils.ClientTestCase, utils.CrudTests): def setUp(self): if oauth1 is None: self.skipTest('oauthlib package not available') super(ConsumerTests, self).setUp() self.key = 'consumer' self.collection_key = 'consumers' self.model = consumers.Consumer self.manager = self.client.oauth1.consumers self.path_prefix = 'OS-OAUTH1' def new_ref(self, **kwargs): kwargs = super(ConsumerTests, self).new_ref(**kwargs) kwargs.setdefault('description', uuid.uuid4().hex) return kwargs def test_description_is_optional(self): consumer_id = uuid.uuid4().hex resp_ref = {'consumer': {'description': None, 'id': consumer_id}} self.stub_url('POST', [self.path_prefix, self.collection_key], status_code=201, json=resp_ref) consumer = self.manager.create() self.assertEqual(consumer_id, consumer.id) self.assertIsNone(consumer.description) def test_description_not_included(self): consumer_id = uuid.uuid4().hex resp_ref = {'consumer': {'id': consumer_id}} self.stub_url('POST', [self.path_prefix, self.collection_key], status_code=201, json=resp_ref) consumer = self.manager.create() self.assertEqual(consumer_id, consumer.id) class TokenTests(object): def _new_oauth_token(self): key = uuid.uuid4().hex secret = uuid.uuid4().hex params = {'oauth_token': key, 'oauth_token_secret': secret} token = urlparse.urlencode(params) return (key, secret, token) def _new_oauth_token_with_expires_at(self): key, secret, token = self._new_oauth_token() expires_at = client_utils.strtime() params = {'oauth_token': key, 'oauth_token_secret': secret, 'oauth_expires_at': expires_at} token = urlparse.urlencode(params) return (key, secret, expires_at, token) def _validate_oauth_headers(self, auth_header, oauth_client): """Validate data in the headers. Assert that the data in the headers matches the data that is produced from oauthlib. """ self.assertThat(auth_header, matchers.StartsWith('OAuth ')) parameters = dict( oauth1.rfc5849.utils.parse_authorization_header(auth_header)) self.assertEqual('HMAC-SHA1', parameters['oauth_signature_method']) self.assertEqual('1.0', parameters['oauth_version']) self.assertIsInstance(parameters['oauth_nonce'], six.string_types) self.assertEqual(oauth_client.client_key, parameters['oauth_consumer_key']) if oauth_client.resource_owner_key: self.assertEqual(oauth_client.resource_owner_key, parameters['oauth_token'],) if oauth_client.verifier: self.assertEqual(oauth_client.verifier, parameters['oauth_verifier']) if oauth_client.callback_uri: self.assertEqual(oauth_client.callback_uri, parameters['oauth_callback']) return parameters class RequestTokenTests(utils.ClientTestCase, TokenTests): def setUp(self): if oauth1 is None: self.skipTest('oauthlib package not available') super(RequestTokenTests, self).setUp() self.model = request_tokens.RequestToken self.manager = self.client.oauth1.request_tokens self.path_prefix = 'OS-OAUTH1' def test_authorize_request_token(self): request_key = uuid.uuid4().hex info = {'id': request_key, 'key': request_key, 'secret': uuid.uuid4().hex} request_token = request_tokens.RequestToken(self.manager, info) verifier = uuid.uuid4().hex resp_ref = {'token': {'oauth_verifier': verifier}} self.stub_url('PUT', [self.path_prefix, 'authorize', request_key], status_code=200, json=resp_ref) # Assert the manager is returning the expected data role_id = uuid.uuid4().hex token = request_token.authorize([role_id]) self.assertEqual(verifier, token.oauth_verifier) # Assert that the request was sent in the expected structure exp_body = {'roles': [{'id': role_id}]} self.assertRequestBodyIs(json=exp_body) def test_create_request_token(self): project_id = uuid.uuid4().hex consumer_key = uuid.uuid4().hex consumer_secret = uuid.uuid4().hex request_key, request_secret, resp_ref = self._new_oauth_token() headers = {'Content-Type': 'application/x-www-form-urlencoded'} self.stub_url('POST', [self.path_prefix, 'request_token'], status_code=201, text=resp_ref, headers=headers) # Assert the manager is returning request token object request_token = self.manager.create(consumer_key, consumer_secret, project_id) self.assertIsInstance(request_token, self.model) self.assertEqual(request_key, request_token.key) self.assertEqual(request_secret, request_token.secret) # Assert that the project id is in the header self.assertRequestHeaderEqual('requested-project-id', project_id) req_headers = self.requests_mock.last_request.headers oauth_client = oauth1.Client(consumer_key, client_secret=consumer_secret, signature_method=oauth1.SIGNATURE_HMAC, callback_uri="oob") self._validate_oauth_headers(req_headers['Authorization'], oauth_client) class AccessTokenTests(utils.ClientTestCase, TokenTests): def setUp(self): if oauth1 is None: self.skipTest('oauthlib package not available') super(AccessTokenTests, self).setUp() self.manager = self.client.oauth1.access_tokens self.model = access_tokens.AccessToken self.path_prefix = 'OS-OAUTH1' def test_create_access_token_expires_at(self): verifier = uuid.uuid4().hex consumer_key = uuid.uuid4().hex consumer_secret = uuid.uuid4().hex request_key = uuid.uuid4().hex request_secret = uuid.uuid4().hex t = self._new_oauth_token_with_expires_at() access_key, access_secret, expires_at, resp_ref = t headers = {'Content-Type': 'application/x-www-form-urlencoded'} self.stub_url('POST', [self.path_prefix, 'access_token'], status_code=201, text=resp_ref, headers=headers) # Assert that the manager creates an access token object access_token = self.manager.create(consumer_key, consumer_secret, request_key, request_secret, verifier) self.assertIsInstance(access_token, self.model) self.assertEqual(access_key, access_token.key) self.assertEqual(access_secret, access_token.secret) self.assertEqual(expires_at, access_token.expires) req_headers = self.requests_mock.last_request.headers oauth_client = oauth1.Client(consumer_key, client_secret=consumer_secret, resource_owner_key=request_key, resource_owner_secret=request_secret, signature_method=oauth1.SIGNATURE_HMAC, verifier=verifier) self._validate_oauth_headers(req_headers['Authorization'], oauth_client) class AuthenticateWithOAuthTests(utils.TestCase, TokenTests): def setUp(self): super(AuthenticateWithOAuthTests, self).setUp() if oauth1 is None: self.skipTest('optional package oauthlib is not installed') def test_oauth_authenticate_success(self): consumer_key = uuid.uuid4().hex consumer_secret = uuid.uuid4().hex access_key = uuid.uuid4().hex access_secret = uuid.uuid4().hex # Just use an existing project scoped token and change # the methods to oauth1, and add an OS-OAUTH1 section. oauth_token = client_fixtures.project_scoped_token() oauth_token['methods'] = ["oauth1"] oauth_token['OS-OAUTH1'] = {"consumer_id": consumer_key, "access_token_id": access_key} self.stub_auth(json=oauth_token) with self.deprecations.expect_deprecations_here(): a = auth.OAuth(self.TEST_URL, consumer_key=consumer_key, consumer_secret=consumer_secret, access_key=access_key, access_secret=access_secret) s = session.Session(auth=a) t = s.get_token() self.assertEqual(self.TEST_TOKEN, t) OAUTH_REQUEST_BODY = { "auth": { "identity": { "methods": ["oauth1"], "oauth1": {} } } } self.assertRequestBodyIs(json=OAUTH_REQUEST_BODY) # Assert that the headers have the same oauthlib data req_headers = self.requests_mock.last_request.headers oauth_client = oauth1.Client(consumer_key, client_secret=consumer_secret, resource_owner_key=access_key, resource_owner_secret=access_secret, signature_method=oauth1.SIGNATURE_HMAC) self._validate_oauth_headers(req_headers['Authorization'], oauth_client) class OauthRequestIdTests(utils.TestRequestId, TokenTests): def setUp(self): super(OauthRequestIdTests, self).setUp() self.mgr = consumers.ConsumerManager(self.client) def _mock_request_method(self, method=None, body=None): return self.useFixture(fixtures.MockPatchObject( self.client, method, autospec=True, return_value=(self.resp, body)) ).mock def test_get_consumers(self): body = {"consumer": {"name": "admin"}} get_mock = self._mock_request_method(method='get', body=body) response = self.mgr.get("admin") self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID) get_mock.assert_called_once_with('/OS-OAUTH1/consumers/admin') def test_create_consumers(self): body = {"consumer": {"name": "admin"}} post_mock = self._mock_request_method(method='post', body=body) response = self.mgr.create(name="admin", description="fake") self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID) post_mock.assert_called_once_with('/OS-OAUTH1/consumers', body={ 'consumer': {'name': 'admin', 'description': 'fake'}}) def test_update_consumers(self): body = {"consumer": {"name": "admin"}} patch_mock = self._mock_request_method(method='patch', body=body) self._mock_request_method(method='post', body=body) response = self.mgr.update("admin", "demo") self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID) patch_mock.assert_called_once_with('/OS-OAUTH1/consumers/admin', body={ 'consumer': {'description': 'demo'}}) def test_delete_consumers(self): get_mock = self._mock_request_method(method='delete') _, resp = self.mgr.delete("admin") self.assertEqual(resp.request_ids[0], self.TEST_REQUEST_ID) get_mock.assert_called_once_with('/OS-OAUTH1/consumers/admin') class TestOAuthLibModule(utils.TestCase): def test_no_oauthlib_installed(self): with mock.patch.object(auth, 'oauth1', None): self.assertRaises(NotImplementedError, auth.OAuth, self.TEST_URL, consumer_key=uuid.uuid4().hex, consumer_secret=uuid.uuid4().hex, access_key=uuid.uuid4().hex, access_secret=uuid.uuid4().hex)