This file is indexed.

/usr/lib/python2.7/dist-packages/keystoneauth1/tests/unit/extras/oauth1/test_oauth1.py is in python-keystoneauth1 3.4.0-0ubuntu1.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
# 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.

import uuid

from oauthlib import oauth1
import six
from testtools import matchers

from keystoneauth1.extras import oauth1 as ksa_oauth1
from keystoneauth1 import fixture
from keystoneauth1 import session
from keystoneauth1.tests.unit import utils as test_utils


class OAuth1AuthTests(test_utils.TestCase):

    TEST_ROOT_URL = 'http://127.0.0.1:5000/'
    TEST_URL = '%s%s' % (TEST_ROOT_URL, 'v3')
    TEST_TOKEN = uuid.uuid4().hex

    def stub_auth(self, subject_token=None, **kwargs):
        if not subject_token:
            subject_token = self.TEST_TOKEN

        self.stub_url('POST', ['auth', 'tokens'],
                      headers={'X-Subject-Token': subject_token}, **kwargs)

    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

    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

        oauth_token = fixture.V3Token(methods=['oauth1'],
                                      oauth_consumer_id=consumer_key,
                                      oauth_access_token_id=access_key)
        oauth_token.set_project_scope()

        self.stub_auth(json=oauth_token)

        a = ksa_oauth1.V3OAuth1(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)

    def test_warning_dual_scope(self):
        ksa_oauth1.V3OAuth1(self.TEST_URL,
                            consumer_key=uuid.uuid4().hex,
                            consumer_secret=uuid.uuid4().hex,
                            access_key=uuid.uuid4().hex,
                            access_secret=uuid.uuid4().hex,
                            project_id=uuid.uuid4().hex)

        self.assertIn('ignored by the identity server', self.logger.output)