4e0846d208
It's tricky to debug authentication when we leave no traces in the logs and the client only receives a generic error. Change-Id: I2c248b94938ec37f4b28e0fda4eded51ee48cdc6
224 lines
8.3 KiB
Python
224 lines
8.3 KiB
Python
# Copyright 2020 Red Hat, Inc.
|
|
# All Rights Reserved.
|
|
#
|
|
# 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 base64
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from unittest import mock
|
|
|
|
from ironic_lib import auth_basic
|
|
from ironic_lib import exception
|
|
from ironic_lib.tests import base
|
|
|
|
|
|
class TestAuthBasic(base.IronicLibTestCase):
|
|
|
|
def write_auth_file(self, data=None):
|
|
if not data:
|
|
data = '\n'
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
|
|
f.write(data)
|
|
self.addCleanup(os.remove, f.name)
|
|
return f.name
|
|
|
|
def test_middleware_authenticate(self):
|
|
auth_file = self.write_auth_file(
|
|
'myName:$2y$05$lE3eGtyj41jZwrzS87KTqe6.'
|
|
'JETVCWBkc32C63UP2aYrGoYOEpbJm\n\n\n')
|
|
app = mock.Mock()
|
|
start_response = mock.Mock()
|
|
middleware = auth_basic.BasicAuthMiddleware(app, auth_file)
|
|
env = {
|
|
'HTTP_AUTHORIZATION': 'Basic bXlOYW1lOm15UGFzc3dvcmQ='
|
|
}
|
|
|
|
result = middleware(env, start_response)
|
|
self.assertEqual(app.return_value, result)
|
|
start_response.assert_not_called()
|
|
|
|
def test_middleware_unauthenticated(self):
|
|
auth_file = self.write_auth_file(
|
|
'myName:$2y$05$lE3eGtyj41jZwrzS87KTqe6.'
|
|
'JETVCWBkc32C63UP2aYrGoYOEpbJm\n\n\n')
|
|
app = mock.Mock()
|
|
start_response = mock.Mock()
|
|
middleware = auth_basic.BasicAuthMiddleware(app, auth_file)
|
|
env = {'REQUEST_METHOD': 'GET'}
|
|
|
|
body = middleware(env, start_response)
|
|
decoded = json.loads(body[0].decode())
|
|
self.assertEqual({'error': {'message': 'Authorization required',
|
|
'code': 401}}, decoded)
|
|
|
|
start_response.assert_called_once_with(
|
|
'401 Unauthorized',
|
|
[('WWW-Authenticate', 'Basic realm="Baremetal API"'),
|
|
('Content-Type', 'application/json'),
|
|
('Content-Length', str(len(body[0])))]
|
|
)
|
|
app.assert_not_called()
|
|
|
|
def test_authenticate(self):
|
|
auth_file = self.write_auth_file(
|
|
'foo:bar\nmyName:$2y$05$lE3eGtyj41jZwrzS87KTqe6.'
|
|
'JETVCWBkc32C63UP2aYrGoYOEpbJm\n\n\n')
|
|
|
|
# test basic auth
|
|
self.assertEqual(
|
|
{'HTTP_X_USER': 'myName', 'HTTP_X_USER_NAME': 'myName'},
|
|
auth_basic.authenticate(
|
|
auth_file, 'myName', b'myPassword')
|
|
)
|
|
|
|
# test failed auth
|
|
e = self.assertRaises(exception.ConfigInvalid,
|
|
auth_basic.authenticate,
|
|
auth_file, 'foo', b'bar')
|
|
self.assertEqual('Invalid configuration file. Only bcrypt digested '
|
|
'passwords are supported for foo', str(e))
|
|
|
|
# test problem reading user data file
|
|
auth_file = auth_file + '.missing'
|
|
e = self.assertRaises(exception.ConfigInvalid,
|
|
auth_basic.authenticate,
|
|
auth_file, 'myName',
|
|
b'myPassword')
|
|
self.assertEqual('Invalid configuration file. Problem reading '
|
|
'auth user file', str(e))
|
|
|
|
def test_auth_entry(self):
|
|
entry_pass = ('myName:$2y$05$lE3eGtyj41jZwrzS87KTqe6.'
|
|
'JETVCWBkc32C63UP2aYrGoYOEpbJm')
|
|
entry_pass_2a = ('myName:$2a$10$I9Fi3DM1sbxQP0560MK9'
|
|
'tec1dUdytBtIqXfDCyTNfDUabtGvQjW1S')
|
|
entry_pass_2b = ('myName:$2b$12$dWLBxT6aMxpVTfUNAyOu'
|
|
'IusHXewu8m6Hrsxw4/e95WGBelFn0oOMW')
|
|
entry_fail = 'foo:bar'
|
|
|
|
# success
|
|
self.assertEqual(
|
|
{'HTTP_X_USER': 'myName', 'HTTP_X_USER_NAME': 'myName'},
|
|
auth_basic.auth_entry(
|
|
entry_pass, b'myPassword')
|
|
)
|
|
|
|
# success with a bcrypt implementations other than htpasswd
|
|
self.assertEqual(
|
|
{'HTTP_X_USER': 'myName', 'HTTP_X_USER_NAME': 'myName'},
|
|
auth_basic.auth_entry(
|
|
entry_pass_2a, b'myPassword')
|
|
)
|
|
self.assertEqual(
|
|
{'HTTP_X_USER': 'myName', 'HTTP_X_USER_NAME': 'myName'},
|
|
auth_basic.auth_entry(
|
|
entry_pass_2b, b'myPassword')
|
|
)
|
|
|
|
# failed, unknown digest format
|
|
e = self.assertRaises(exception.ConfigInvalid,
|
|
auth_basic.auth_entry, entry_fail, b'bar')
|
|
self.assertEqual('Invalid configuration file. Only bcrypt digested '
|
|
'passwords are supported for foo', str(e))
|
|
|
|
# failed, incorrect password
|
|
e = self.assertRaises(exception.Unauthorized,
|
|
auth_basic.auth_entry, entry_pass, b'bar')
|
|
self.assertEqual('Incorrect username or password', str(e))
|
|
|
|
def test_validate_auth_file(self):
|
|
auth_file = self.write_auth_file(
|
|
'myName:$2y$05$lE3eGtyj41jZwrzS87KTqe6.'
|
|
'JETVCWBkc32C63UP2aYrGoYOEpbJm\n\n\n')
|
|
# success, valid config
|
|
auth_basic.validate_auth_file(auth_file)
|
|
|
|
# failed, missing auth file
|
|
auth_file = auth_file + '.missing'
|
|
self.assertRaises(exception.ConfigInvalid,
|
|
auth_basic.validate_auth_file, auth_file)
|
|
|
|
# failed, invalid entry
|
|
auth_file = self.write_auth_file(
|
|
'foo:bar\nmyName:$2y$05$lE3eGtyj41jZwrzS87KTqe6.'
|
|
'JETVCWBkc32C63UP2aYrGoYOEpbJm\n\n\n')
|
|
self.assertRaises(exception.ConfigInvalid,
|
|
auth_basic.validate_auth_file, auth_file)
|
|
|
|
def test_parse_token(self):
|
|
|
|
# success with bytes
|
|
token = base64.b64encode(b'myName:myPassword')
|
|
self.assertEqual(
|
|
('myName', b'myPassword'),
|
|
auth_basic.parse_token(token)
|
|
)
|
|
|
|
# success with string
|
|
token = str(token, encoding='utf-8')
|
|
self.assertEqual(
|
|
('myName', b'myPassword'),
|
|
auth_basic.parse_token(token)
|
|
)
|
|
|
|
# failed, invalid base64
|
|
e = self.assertRaises(exception.BadRequest,
|
|
auth_basic.parse_token, token[:-1])
|
|
self.assertEqual('Could not decode authorization token', str(e))
|
|
|
|
# failed, no colon in token
|
|
token = str(base64.b64encode(b'myNamemyPassword'), encoding='utf-8')
|
|
e = self.assertRaises(exception.BadRequest,
|
|
auth_basic.parse_token, token[:-1])
|
|
self.assertEqual('Could not decode authorization token', str(e))
|
|
|
|
def test_parse_header(self):
|
|
auth_value = 'Basic bXlOYW1lOm15UGFzc3dvcmQ='
|
|
|
|
# success
|
|
self.assertEqual(
|
|
'bXlOYW1lOm15UGFzc3dvcmQ=',
|
|
auth_basic.parse_header({
|
|
'HTTP_AUTHORIZATION': auth_value
|
|
})
|
|
)
|
|
|
|
# failed, missing Authorization header
|
|
e = self.assertRaises(exception.Unauthorized,
|
|
auth_basic.parse_header,
|
|
{})
|
|
self.assertEqual('Authorization required', str(e))
|
|
|
|
# failed missing token
|
|
e = self.assertRaises(exception.BadRequest,
|
|
auth_basic.parse_header,
|
|
{'HTTP_AUTHORIZATION': 'Basic'})
|
|
self.assertEqual('Could not parse Authorization header', str(e))
|
|
|
|
# failed, type other than Basic
|
|
digest_value = 'Digest username="myName" nonce="foobar"'
|
|
e = self.assertRaises(exception.BadRequest,
|
|
auth_basic.parse_header,
|
|
{'HTTP_AUTHORIZATION': digest_value})
|
|
self.assertEqual('Unsupported authorization type "Digest"', str(e))
|
|
|
|
def test_unauthorized(self):
|
|
e = self.assertRaises(exception.Unauthorized,
|
|
auth_basic.unauthorized, 'ouch')
|
|
self.assertEqual('ouch', str(e))
|
|
self.assertEqual({
|
|
'WWW-Authenticate': 'Basic realm="Baremetal API"'
|
|
}, e.headers)
|