oslo.middleware/oslo_middleware/basic_auth.py

# Copyright 2012 OpenStack Foundation
# 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 oslo_config import cfg
from oslo_middleware import base

LOG = logging.getLogger(__name__)

OPTS = [
    cfg.StrOpt('http_basic_auth_user_file',
               default='/etc/htpasswd',
               help="HTTP basic auth password file.")
]

cfg.CONF.register_opts(OPTS, group='oslo_middleware')


class ConfigInvalid(Exception):
    def __init__(self, error_msg):
        super().__init__(
            'Invalid configuration file. %(error_msg)s')


class BasicAuthMiddleware(base.ConfigurableMiddleware):
    """Middleware which performs HTTP basic authentication on requests"""

    def __init__(self, application, conf=None):
        super().__init__(application, conf)
        self.auth_file = cfg.CONF.oslo_middleware.http_basic_auth_user_file
        validate_auth_file(self.auth_file)

    def format_exception(self, e):
        result = {'error': {'message': str(e), 'code': 401}}
        headers = [('Content-Type', 'application/json')]
        return webob.Response(content_type='application/json',
                              status_code=401,
                              json_body=result,
                              headerlist=headers)

    @webob.dec.wsgify
    def __call__(self, req):
        try:
            token = parse_header(req.environ)
            username, password = parse_token(token)
            req.environ.update(authenticate(
                self.auth_file, username, password))
            return self.application
        except Exception as e:
            response = self.format_exception(e)
            return self.process_response(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: HTTPUnauthorized, if no file entries match 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 file: %s', exc)
        raise webob.exc.HTTPBadRequest(
            detail='Problem reading auth file')
    # reached end of file with no matches
    LOG.info('User %s not found', username)
    raise webob.exc.HTTPUnauthorized()


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: HTTPUnauthorized, if the entry doesn't match supplied password or
        if the entry is crypted with a method other than bcrypt
    """

    username, crypted = parse_entry(entry)
    if not bcrypt.checkpw(password, crypted):
        LOG.info('Password for %s does not match', username)
        raise webob.exc.HTTPUnauthorized()
    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 ConfigInvalid(error_msg='Problem reading auth user file')


def parse_entry(entry):
    """Extrace the username and crypted password from a user auth file entry

    :param: entry: Line from auth user file to use for authentication
    :returns: a tuple of username and crypted password
    :raises: ConfigInvalid if the password is not in the supported bcrypt
    format
    """

    username, crypted_str = entry.split(':', maxsplit=1)
    crypted = crypted_str.encode('utf-8')
    if crypted[:4] not in (b'$2y$', b'$2a$', b'$2b$'):
        error_msg = ('Only bcrypt digested passwords are supported for '
                     '%(username)s') % {'username': username}
        raise webob.exc.HTTPBadRequest(detail=error_msg)
    return username, crypted


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: BadRequest, 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 webob.exc.HTTPBadRequest(detail=(
            '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: HTTPUnauthorized, 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')
        raise webob.exc.HTTPUnauthorized()
    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 webob.exc.HTTPBadRequest(detail=(
            'Could not parse Authorization header'))
    if auth_type.lower() != 'basic':
        error_msg = ('Unsupported authorization type "%s"') % auth_type
        LOG.info(error_msg)
        raise webob.exc.HTTPBadRequest(detail=error_msg)
    return token