374 lines
14 KiB
Python
374 lines
14 KiB
Python
# Copyright 2022 OpenStack Foundation
|
|
#
|
|
# 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 json
|
|
import logging
|
|
import ssl
|
|
from unittest import mock
|
|
import uuid
|
|
import webob.dec
|
|
|
|
import fixtures
|
|
from oslo_config import cfg
|
|
import six
|
|
from six.moves import http_client
|
|
import testresources
|
|
|
|
from keystoneauth1 import access
|
|
from keystoneauth1 import exceptions as ksa_exceptions
|
|
|
|
from keystonemiddleware import oauth2_mtls_token
|
|
from keystonemiddleware.tests.unit.auth_token import base
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import BASE_URI
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import ENDPOINT_NOT_FOUND_TOKEN
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import ERROR_TOKEN
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import FAKE_ADMIN_TOKEN
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import FAKE_ADMIN_TOKEN_ID
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import FakeApp
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import FakeOsloCache
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import request_timeout_response
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import TIMEOUT_TOKEN
|
|
from keystonemiddleware.tests.unit.auth_token.test_auth_token_middleware \
|
|
import VERSION_LIST_v3
|
|
from keystonemiddleware.tests.unit import client_fixtures
|
|
from keystonemiddleware.tests.unit.test_oauth2_token_middleware \
|
|
import FakeOauth2TokenV3App
|
|
from keystonemiddleware.tests.unit.test_oauth2_token_middleware \
|
|
import get_authorization_header
|
|
from keystonemiddleware.tests.unit import utils
|
|
|
|
_no_value = object()
|
|
|
|
|
|
class FakeSocket(object):
|
|
|
|
def __init__(self, binary_peer_cert):
|
|
self.binary_peer_cert = binary_peer_cert
|
|
|
|
def getpeercert(self, binary_form=True):
|
|
return self.binary_peer_cert
|
|
|
|
|
|
class FakeWsgiInput(object):
|
|
|
|
def __init__(self, fake_socket):
|
|
self.fake_socket = fake_socket
|
|
|
|
def get_socket(self):
|
|
return self.fake_socket
|
|
|
|
|
|
class BaseOauth2mTlsTokenMiddlewareTest(base.BaseAuthTokenTestCase):
|
|
|
|
def setUp(self, expected_env=None, auth_version=None, fake_app=None):
|
|
cfg.CONF.clear()
|
|
super(BaseOauth2mTlsTokenMiddlewareTest, self).setUp()
|
|
|
|
self.logger = self.useFixture(fixtures.FakeLogger(level=logging.DEBUG))
|
|
self.useFixture(
|
|
fixtures.MockPatchObject(oauth2_mtls_token.OAuth2mTlsProtocol,
|
|
'_create_oslo_cache',
|
|
return_value=FakeOsloCache))
|
|
self.expected_env = expected_env or dict()
|
|
self.fake_app = fake_app or FakeApp
|
|
self.middleware = None
|
|
|
|
self.conf = {
|
|
'identity_uri': 'https://keystone.example.com:1234/testadmin/',
|
|
'auth_version': auth_version,
|
|
'www_authenticate_uri': 'https://keystone.example.com:1234',
|
|
'admin_user': uuid.uuid4().hex,
|
|
}
|
|
self.auth_version = auth_version
|
|
|
|
def call_middleware(self, **kwargs):
|
|
return self.call(self.middleware, **kwargs)
|
|
|
|
def set_middleware(self, expected_env=None, conf=None):
|
|
"""Configure the class ready to call the oauth2_token middleware.
|
|
|
|
Set up the various fake items needed to run the middleware.
|
|
Individual tests that need to further refine these can call this
|
|
function to override the class defaults.
|
|
|
|
"""
|
|
if conf:
|
|
self.conf.update(conf)
|
|
|
|
if expected_env:
|
|
self.expected_env.update(expected_env)
|
|
self.middleware = oauth2_mtls_token.OAuth2mTlsProtocol(
|
|
self.fake_app(self.expected_env), self.conf)
|
|
|
|
def call(self, middleware, method='GET', path='/', headers=None,
|
|
expected_status=http_client.OK,
|
|
expected_body_string=None, **kwargs):
|
|
req = webob.Request.blank(path, **kwargs)
|
|
req.method = method
|
|
|
|
for k, v in (headers or {}).items():
|
|
req.headers[k] = v
|
|
|
|
resp = req.get_response(middleware)
|
|
self.assertEqual(expected_status, resp.status_int)
|
|
if expected_body_string:
|
|
self.assertIn(expected_body_string, six.text_type(resp.body))
|
|
resp.request = req
|
|
return resp
|
|
|
|
def assertUnauthorizedResp(self, resp):
|
|
error = json.loads(resp.body)
|
|
|
|
self.assertEqual('Keystone uri="https://keystone.example.com:1234"',
|
|
resp.headers['WWW-Authenticate'])
|
|
self.assertEqual(
|
|
'Keystone uri="%s"' % self.conf.get('www_authenticate_uri'),
|
|
resp.headers['WWW-Authenticate'])
|
|
self.assertEqual(
|
|
'Unauthorized',
|
|
error.get('error').get('title'))
|
|
self.assertEqual(
|
|
'The request you have made requires authentication.',
|
|
error.get('error').get('message'))
|
|
|
|
|
|
class Oauth2mTlsTokenMiddlewareTest(BaseOauth2mTlsTokenMiddlewareTest,
|
|
testresources.ResourcedTestCase):
|
|
resources = [('examples', client_fixtures.EXAMPLES_RESOURCE)]
|
|
|
|
def setUp(self):
|
|
super(Oauth2mTlsTokenMiddlewareTest, self).setUp(
|
|
auth_version='v3.0',
|
|
fake_app=FakeOauth2TokenV3App)
|
|
self.requests_mock.post('%s/v2.0/tokens' % BASE_URI,
|
|
text=FAKE_ADMIN_TOKEN)
|
|
self.requests_mock.get(BASE_URI,
|
|
json=VERSION_LIST_v3,
|
|
status_code=300)
|
|
self.requests_mock.get('%s/v3/auth/tokens' % BASE_URI,
|
|
text=self.token_response,
|
|
headers={'X-Subject-Token': uuid.uuid4().hex})
|
|
self.set_middleware(conf={'service_type': 'tacker'})
|
|
|
|
def token_response(self, request, context):
|
|
auth_id = request.headers.get('X-Auth-Token')
|
|
token_id = request.headers.get('X-Subject-Token')
|
|
self.assertEqual(auth_id, FAKE_ADMIN_TOKEN_ID)
|
|
|
|
if token_id == ERROR_TOKEN:
|
|
msg = "Network connection refused."
|
|
raise ksa_exceptions.ConnectFailure(msg)
|
|
if token_id == ENDPOINT_NOT_FOUND_TOKEN:
|
|
raise ksa_exceptions.EndpointNotFound()
|
|
if token_id == TIMEOUT_TOKEN:
|
|
request_timeout_response(request, context)
|
|
|
|
try:
|
|
response = self.examples.JSON_TOKEN_RESPONSES[token_id]
|
|
except KeyError:
|
|
response = ""
|
|
context.status_code = 404
|
|
|
|
return response
|
|
|
|
def call_middleware(self, client_cert=_no_value, **kwargs):
|
|
if client_cert is _no_value:
|
|
client_cert = self.examples.V3_OAUTH2_MTLS_CERTIFICATE
|
|
|
|
if client_cert:
|
|
fake_socket = FakeSocket(client_cert)
|
|
fake_wsgi_input = FakeWsgiInput(fake_socket)
|
|
kwargs.update({'environ': {'wsgi.input': fake_wsgi_input}})
|
|
|
|
return self.call(self.middleware, **kwargs)
|
|
|
|
def test_basic(self):
|
|
token = self.examples.v3_OAUTH2_CREDENTIAL
|
|
token_data = self.examples.TOKEN_RESPONSES[token]
|
|
|
|
resp = self.call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=200,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages',
|
|
)
|
|
self.assertEqual(FakeApp.SUCCESS, resp.body)
|
|
token_auth = resp.request.environ['keystone.token_auth']
|
|
self.assertTrue(token_auth.has_user_token)
|
|
self.assertEqual(token_data.user_id, token_auth.user.user_id)
|
|
self.assertEqual(token_data.project_id, token_auth.user.project_id)
|
|
self.assertEqual(token_data.user_domain_id,
|
|
token_auth.user.user_domain_id)
|
|
self.assertEqual(token_data.project_domain_id,
|
|
token_auth.user.project_domain_id)
|
|
self.assertEqual(token_data.oauth2_thumbprint,
|
|
token_auth.user.oauth2_credential_thumbprint)
|
|
|
|
def test_not_oauth2_credential_token(self):
|
|
token = self.examples.v3_APP_CRED_TOKEN
|
|
resp = self.call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=401,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages',
|
|
)
|
|
self.assertUnauthorizedResp(resp)
|
|
self.assertIn(
|
|
'Invalid OAuth2.0 certificate-bound access token: '
|
|
'The token is not an OAuth2.0 credential access token.',
|
|
self.logger.output)
|
|
|
|
def test_thumbprint_not_match(self):
|
|
diff_cert = self.examples.V3_OAUTH2_MTLS_CERTIFICATE_DIFF
|
|
token = self.examples.v3_OAUTH2_CREDENTIAL
|
|
resp = self.call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=401,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages',
|
|
client_cert=diff_cert
|
|
)
|
|
self.assertUnauthorizedResp(resp)
|
|
self.assertIn('The two thumbprints do not match.',
|
|
self.logger.output)
|
|
|
|
@mock.patch.object(ssl, 'DER_cert_to_PEM_cert')
|
|
def test_gen_thumbprint_exception(self, mock_DER_cert_to_PEM_cert):
|
|
except_msg = 'Boom!'
|
|
mock_DER_cert_to_PEM_cert.side_effect = Exception(except_msg)
|
|
token = self.examples.v3_OAUTH2_CREDENTIAL
|
|
resp = self.call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=401,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages'
|
|
)
|
|
self.assertUnauthorizedResp(resp)
|
|
self.assertIn(except_msg, self.logger.output)
|
|
|
|
def test_without_cert(self):
|
|
token = self.examples.v3_OAUTH2_CREDENTIAL
|
|
resp = self.call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=401,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages',
|
|
client_cert=None
|
|
)
|
|
self.assertUnauthorizedResp(resp)
|
|
self.assertIn('Unable to obtain the client certificate.',
|
|
self.logger.output)
|
|
|
|
def test_not_wsgi_input(self):
|
|
token = self.examples.v3_OAUTH2_CREDENTIAL
|
|
resp = super(Oauth2mTlsTokenMiddlewareTest, self).call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=401,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages',
|
|
environ={'wsgi.input': None}
|
|
)
|
|
self.assertUnauthorizedResp(resp)
|
|
self.assertIn('Unable to obtain the client certificate.',
|
|
self.logger.output)
|
|
|
|
def test_not_socket(self):
|
|
token = self.examples.v3_OAUTH2_CREDENTIAL
|
|
resp = super(Oauth2mTlsTokenMiddlewareTest, self).call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=401,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages',
|
|
environ={'wsgi.input': FakeWsgiInput(None)}
|
|
)
|
|
self.assertUnauthorizedResp(resp)
|
|
self.assertIn('Unable to obtain the client certificate.',
|
|
self.logger.output)
|
|
|
|
def test_not_peer_cert(self):
|
|
token = self.examples.v3_OAUTH2_CREDENTIAL
|
|
resp = super(Oauth2mTlsTokenMiddlewareTest, self).call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=401,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages',
|
|
environ={'wsgi.input': FakeWsgiInput(FakeSocket(None))}
|
|
)
|
|
self.assertUnauthorizedResp(resp)
|
|
self.assertIn('Unable to obtain the client certificate.',
|
|
self.logger.output)
|
|
|
|
@mock.patch.object(access, 'create')
|
|
def test_keystonemiddleware_exceptiton(self, mock_create):
|
|
except_msg = 'Unrecognized auth response'
|
|
mock_create.side_effect = Exception(except_msg)
|
|
token = self.examples.v3_OAUTH2_CREDENTIAL
|
|
resp = self.call_middleware(
|
|
headers=get_authorization_header(token),
|
|
expected_status=401,
|
|
method='GET', path='/vnfpkgm/v1/vnf_packages',
|
|
)
|
|
self.assertUnauthorizedResp(resp)
|
|
self.assertIn(
|
|
'Invalid token contents.',
|
|
self.logger.output)
|
|
self.assertIn(
|
|
'Invalid OAuth2.0 certificate-bound access token: %s'
|
|
% 'Token authorization failed',
|
|
self.logger.output)
|
|
|
|
def test_request_no_token(self):
|
|
resp = self.call_middleware(expected_status=401)
|
|
self.assertUnauthorizedResp(resp)
|
|
|
|
def test_request_blank_token(self):
|
|
resp = self.call_middleware(headers=get_authorization_header(''),
|
|
expected_status=401)
|
|
self.assertUnauthorizedResp(resp)
|
|
|
|
def _get_cached_token(self, token):
|
|
return self.middleware._token_cache.get(token)
|
|
|
|
def assert_valid_last_url(self, token_id):
|
|
self.assertLastPath('/v3/auth/tokens')
|
|
|
|
def assertLastPath(self, path):
|
|
if path:
|
|
self.assertEqual(BASE_URI + path,
|
|
self.requests_mock.last_request.url)
|
|
else:
|
|
self.assertIsNone(self.requests_mock.last_request)
|
|
|
|
def test_http_error_not_cached_token(self):
|
|
"""Test to don't cache token as invalid on network errors.
|
|
|
|
We use UUID tokens since they are the easiest one to reach
|
|
get_http_connection.
|
|
"""
|
|
self.set_middleware(conf={'http_request_max_retries': '0'})
|
|
self.call_middleware(headers=get_authorization_header(ERROR_TOKEN),
|
|
expected_status=503)
|
|
self.assertIsNone(self._get_cached_token(ERROR_TOKEN))
|
|
self.assert_valid_last_url(ERROR_TOKEN)
|
|
|
|
|
|
class FilterFactoryTest(utils.BaseTestCase):
|
|
|
|
def test_filter_factory(self):
|
|
conf = {}
|
|
auth_filter = oauth2_mtls_token.filter_factory(conf)
|
|
m = auth_filter(FakeOauth2TokenV3App())
|
|
self.assertIsInstance(m, oauth2_mtls_token.OAuth2mTlsProtocol)
|