Import needed ironic-lib code

Ironic-lib is being deprecated; we need to import the auth_basic module
sushy-tools uses.

Change-Id: I333b737f10fa2de81c32537e562af16f274ebe3f
This commit is contained in:
Jay Faulkner 2025-01-15 15:07:44 -08:00
parent bcd694e90c
commit ef58421ba8
5 changed files with 443 additions and 2 deletions

@ -10,4 +10,5 @@ pbr>=6.0.0 # Apache-2.0
Flask>=1.0.2 # BSD
requests>=2.14.2 # Apache-2.0
tenacity>=6.2.0 # Apache-2.0
ironic-lib>=4.6.1 # Apache-2.0
bcrypt>=3.1.3 # Apache-2.0
WebOb>=1.7.1 # MIT

@ -0,0 +1,201 @@
# 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 binascii
import logging
import bcrypt
import webob
from sushy_tools import error
LOG = logging.getLogger(__name__)
class BasicAuthMiddleware(object):
"""Middleware which performs HTTP basic authentication on requests
"""
def __init__(self, app, auth_file):
self.app = app
self.auth_file = auth_file
validate_auth_file(auth_file)
def format_exception(self, e):
result = {'error': {'message': str(e), 'code': e.code}}
headers = list(e.headers.items()) + [
('Content-Type', 'application/json')
]
return webob.Response(content_type='application/json',
status_code=e.code,
json_body=result,
headerlist=headers)
def __call__(self, env, start_response):
try:
token = parse_header(env)
username, password = parse_token(token)
env.update(authenticate(self.auth_file, username, password))
return self.app(env, start_response)
except error.FishyError as e:
response = self.format_exception(e)
return response(env, start_response)
def authenticate(auth_file, username, password):
"""Finds username and password match in Apache style user auth file
The user auth file format is expected to comply with Apache
documentation[1] however the bcrypt password digest is the *only*
digest format supported.
[1] https://httpd.apache.org/docs/current/misc/password_encryptions.html
:param: auth_file: Path to user auth file
:param: username: Username to authenticate
:param: password: Password encoded as bytes
:returns: A dictionary of WSGI environment values to append to the request
:raises: Unauthorized, if no file entries match supplied username/password
"""
line_prefix = username + ':'
try:
with open(auth_file, 'r') as f:
for line in f:
entry = line.strip()
if entry and entry.startswith(line_prefix):
return auth_entry(entry, password)
except OSError as exc:
LOG.error('Problem reading auth user file: %s', exc)
raise error.ConfigInvalid(msg='Problem reading auth user file')
# reached end of file with no matches
LOG.info('User %s not found', username)
unauthorized()
def auth_entry(entry, password):
"""Compare a password with a single user auth file entry
:param: entry: Line from auth user file to use for authentication
:param: password: Password encoded as bytes
:returns: A dictionary of WSGI environment values to append to the request
:raises: Unauthorized, if the entry doesn't match supplied password or
if the entry is encrypted with a method other than bcrypt
"""
username, encrypted = parse_entry(entry)
if not bcrypt.checkpw(password, encrypted):
LOG.info('Password for %s does not match', username)
unauthorized()
return {
'HTTP_X_USER': username,
'HTTP_X_USER_NAME': username
}
def validate_auth_file(auth_file):
"""Read the auth user file and validate its correctness
:param: auth_file: Path to user auth file
:raises: ConfigInvalid on validation error
"""
try:
with open(auth_file, 'r') as f:
for line in f:
entry = line.strip()
if entry and ':' in entry:
parse_entry(entry)
except OSError:
raise error.ConfigInvalid(
msg='Problem reading auth user file: %s' % auth_file)
def parse_entry(entry):
"""Extrace the username and encrypted password from a user auth file entry
:param: entry: Line from auth user file to use for authentication
:returns: a tuple of username and encrypted password
:raises: ConfigInvalid if the password is not in the supported bcrypt
format
"""
username, encrypted_str = entry.split(':', maxsplit=1)
encrypted = encrypted_str.encode('utf-8')
if encrypted[:4] not in (b'$2y$', b'$2a$', b'$2b$'):
error_msg = ('Only bcrypt digested passwords are supported for '
'%(username)s') % {'username': username}
raise error.ConfigInvalid(msg=error_msg)
return username, encrypted
def parse_token(token):
"""Parse the token portion of the Authentication header value
:param: token: Token value from basic authorization header
:returns: tuple of username, password
:raises: Unauthorized, if username and password could not be parsed for any
reason
"""
try:
if isinstance(token, str):
token = token.encode('utf-8')
auth_pair = base64.b64decode(token, validate=True)
(username, password) = auth_pair.split(b':', maxsplit=1)
return (username.decode('utf-8'), password)
except (TypeError, binascii.Error, ValueError) as exc:
LOG.info('Could not decode authorization token: %s', exc)
raise error.BadRequest('Could not decode authorization token')
def parse_header(env):
"""Parse WSGI environment for Authorization header of type Basic
:param: env: WSGI environment to get header from
:returns: Token portion of the header value
:raises: Unauthorized, if header is missing or if the type is not Basic
"""
try:
auth_header = env.pop('HTTP_AUTHORIZATION')
except KeyError:
LOG.info('No authorization token received')
unauthorized('Authorization required')
try:
auth_type, token = auth_header.strip().split(maxsplit=1)
except (ValueError, AttributeError) as exc:
LOG.info('Could not parse Authorization header: %s', exc)
raise error.BadRequest('Could not parse Authorization header')
if auth_type.lower() != 'basic':
msg = ('Unsupported authorization type "%s"') % auth_type
LOG.info(msg)
raise error.BadRequest(msg)
return token
def unauthorized(message=None):
"""Raise an Unauthorized exception to prompt for basic authentication
:param: message: Optional message for esception
:raises: Unauthorized with WWW-Authenticate header set
"""
if not message:
message = 'Incorrect username or password'
raise error.Unauthorized(message)

@ -21,10 +21,10 @@ import ssl
import sys
import flask
from ironic_lib import auth_basic
from werkzeug import exceptions as wz_exc
from sushy_tools.emulator import api_utils
from sushy_tools.emulator import auth_basic
from sushy_tools.emulator.controllers import certificate_service as certctl
from sushy_tools.emulator.controllers import update_service as usctl
from sushy_tools.emulator.controllers import virtual_media as vmctl

@ -59,3 +59,19 @@ class Conflict(FishyError):
def __init__(self, msg, code=409):
super().__init__(msg, code)
class ConfigInvalid(FishyError):
"""Config is invalid."""
def __init__(self, msg, code=500):
errmsg = f"Invalid configuration file. {msg}"
super().__init__(errmsg, code)
class Unauthorized(FishyError):
"""Unauthorized for resource"""
def __init__(self, msg, code=401):
self.headers = {'WWW-Authenticate': 'Basic realm="Baremetal API"'}
super().__init__(msg, code)

@ -0,0 +1,223 @@
# 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 sushy_tools.emulator import auth_basic
from sushy_tools import error
from sushy_tools.tests.unit import base
class TestAuthBasic(base.TestCase):
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(error.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(error.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(error.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(error.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(error.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(error.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(error.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(error.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(error.Unauthorized,
auth_basic.parse_header,
{})
self.assertEqual('Authorization required', str(e))
# failed missing token
e = self.assertRaises(error.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(error.BadRequest,
auth_basic.parse_header,
{'HTTP_AUTHORIZATION': digest_value})
self.assertEqual('Unsupported authorization type "Digest"', str(e))
def test_unauthorized(self):
e = self.assertRaises(error.Unauthorized,
auth_basic.unauthorized, 'ouch')
self.assertEqual('ouch', str(e))
self.assertEqual({
'WWW-Authenticate': 'Basic realm="Baremetal API"'
}, e.headers)