python3-swiftclient 1:3.5.0-0ubuntu1 / usr / lib / python3 / dist-packages / swiftclient / authv1.py

# Copyright 2016 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.

"""
Authentication plugin for keystoneauth to support v1 endpoints.

Way back in the long-long ago, there was no Keystone. Swift used an auth
mechanism now known as "v1", which used only HTTP headers. Auth requests
and responses would look something like::

   > GET /auth/v1.0 HTTP/1.1
   > Host: <swift server>
   > X-Auth-User: <tenant>:<user>
   > X-Auth-Key: <password>
   >
   < HTTP/1.1 200 OK
   < X-Storage-Url: http://<swift server>/v1/<tenant account>
   < X-Auth-Token: <token>
   < X-Storage-Token: <token>
   <

This plugin provides a way for Keystone sessions (and clients that
use them, like python-openstackclient) to communicate with old auth
endpoints that still use this mechanism, such as tempauth, swauth,
or https://identity.api.rackspacecloud.com/v1.0
"""

import datetime
import json
import time

from six.moves.urllib.parse import urljoin

# Note that while we import keystoneauth1 here, we *don't* need to add it to
# requirements.txt -- this entire module only makes sense (and should only be
# loaded) if keystoneauth is already installed.
from keystoneauth1 import plugin
from keystoneauth1 import exceptions
from keystoneauth1 import loading
from keystoneauth1.identity import base


# stupid stdlib...
class _UTC(datetime.tzinfo):
    def utcoffset(self, dt):
        return datetime.timedelta(0)

    def tzname(self, dt):
        return "UTC"

    def dst(self, dt):
        return datetime.timedelta(0)


UTC = _UTC()
del _UTC


class ServiceCatalogV1(object):
    def __init__(self, auth_url, storage_url, account):
        self.auth_url = auth_url
        self._storage_url = storage_url
        self._account = account

    @property
    def storage_url(self):
        if self._account:
            return urljoin(self._storage_url.rstrip('/'), self._account)
        return self._storage_url

    @property
    def catalog(self):
        # openstackclient wants this for the `catalog list` and
        # `catalog show` commands
        endpoints = [{
            'region': 'default',
            'publicURL': self._storage_url,
        }]
        if self.storage_url != self._storage_url:
            endpoints.insert(0, {
                'region': 'override',
                'publicURL': self.storage_url,
            })

        return [
            {
                'name': 'swift',
                'type': 'object-store',
                'endpoints': endpoints,
            },
            {
                'name': 'auth',
                'type': 'identity',
                'endpoints': [{
                    'region': 'default',
                    'publicURL': self.auth_url,
                }],
            }
        ]

    def url_for(self, **kwargs):
        kwargs.setdefault('interface', 'public')
        kwargs.setdefault('service_type', None)

        if kwargs['service_type'] == 'object-store':
            return self.storage_url

        # Although our "catalog" includes an identity entry, nothing that uses
        # url_for() (including `openstack endpoint list`) will know what to do
        # with it. Better to just raise the exception, cribbing error messages
        # from keystoneauth1/access/service_catalog.py

        if 'service_name' in kwargs and 'region_name' in kwargs:
            msg = ('%(interface)s endpoint for %(service_type)s service '
                   'named %(service_name)s in %(region_name)s region not '
                   'found' % kwargs)
        elif 'service_name' in kwargs:
            msg = ('%(interface)s endpoint for %(service_type)s service '
                   'named %(service_name)s not found' % kwargs)
        elif 'region_name' in kwargs:
            msg = ('%(interface)s endpoint for %(service_type)s service '
                   'in %(region_name)s region not found' % kwargs)
        else:
            msg = ('%(interface)s endpoint for %(service_type)s service '
                   'not found' % kwargs)

        raise exceptions.EndpointNotFound(msg)


class AccessInfoV1(object):
    """An object for encapsulating a raw v1 auth token."""

    def __init__(self, auth_url, storage_url, account, username, auth_token,
                 token_life):
        self.auth_url = auth_url
        self.storage_url = storage_url
        self.account = account
        self.service_catalog = ServiceCatalogV1(auth_url, storage_url, account)
        self.username = username
        self.auth_token = auth_token
        self._issued = time.time()
        try:
            self._expires = self._issued + float(token_life)
        except (TypeError, ValueError):
            self._expires = None
        # following is used by openstackclient
        self.project_id = None

    @property
    def expires(self):
        if self._expires is None:
            return None
        return datetime.datetime.fromtimestamp(self._expires, UTC)

    @property
    def issued(self):
        return datetime.datetime.fromtimestamp(self._issued, UTC)

    @property
    def user_id(self):
        # openstackclient wants this for the `token issue` command
        return self.username

    def will_expire_soon(self, stale_duration):
        """Determines if expiration is about to occur.

        :returns: true if expiration is within the given duration
        """
        if self._expires is None:
            return False  # assume no expiration
        return time.time() + stale_duration > self._expires

    def get_state(self):
        """Serialize the current state."""
        return json.dumps({
            'auth_url': self.auth_url,
            'storage_url': self.storage_url,
            'account': self.account,
            'username': self.username,
            'auth_token': self.auth_token,
            'issued': self._issued,
            'expires': self._expires}, sort_keys=True)

    @classmethod
    def from_state(cls, data):
        """Deserialize the given state.

        :returns: a new AccessInfoV1 object with the given state
        """
        data = json.loads(data)
        access = cls(
            data['auth_url'],
            data['storage_url'],
            data['account'],
            data['username'],
            data['auth_token'],
            token_life=None)
        access._issued = data['issued']
        access._expires = data['expires']
        return access


class PasswordPlugin(base.BaseIdentityPlugin):
    """A plugin for authenticating with a username and password.

    Subclassing from BaseIdentityPlugin gets us a few niceties, like handling
    token invalidation and locking during authentication.

    :param string auth_url: Identity v1 endpoint for authorization.
    :param string username: Username for authentication.
    :param string password: Password for authentication.
    :param string project_name: Swift account to use after authentication.
                                We use 'project_name' to be consistent with
                                other auth plugins.
    :param string reauthenticate: Whether to allow re-authentication.
    """
    access_class = AccessInfoV1

    def __init__(self, auth_url, username, password, project_name=None,
                 reauthenticate=True):
        super(PasswordPlugin, self).__init__(
            auth_url=auth_url,
            reauthenticate=reauthenticate)
        self.user = username
        self.key = password
        self.account = project_name

    def get_auth_ref(self, session, **kwargs):
        """Obtain a token from a v1 endpoint.

        This function should not be called independently and is expected to be
        invoked via the do_authenticate function.

        This function will be invoked if the AcessInfo object cached by the
        plugin is not valid. Thus plugins should always fetch a new AccessInfo
        when invoked. If you are looking to just retrieve the current auth
        data then you should use get_access.

        :param session: A session object that can be used for communication.

        :returns: Token access information.
        """
        headers = {'X-Auth-User': self.user,
                   'X-Auth-Key': self.key}

        resp = session.get(self.auth_url, headers=headers,
                           authenticated=False, log=False)

        if resp.status_code // 100 != 2:
            raise exceptions.InvalidResponse(response=resp)

        if 'X-Storage-Url' not in resp.headers:
            raise exceptions.InvalidResponse(response=resp)

        if 'X-Auth-Token' not in resp.headers and \
                'X-Storage-Token' not in resp.headers:
            raise exceptions.InvalidResponse(response=resp)
        token = resp.headers.get('X-Storage-Token',
                                 resp.headers.get('X-Auth-Token'))
        return AccessInfoV1(
            auth_url=self.auth_url,
            storage_url=resp.headers['X-Storage-Url'],
            account=self.account,
            username=self.user,
            auth_token=token,
            token_life=resp.headers.get('X-Auth-Token-Expires'))

    def get_cache_id_elements(self):
        """Get the elements for this auth plugin that make it unique."""
        return {'auth_url': self.auth_url,
                'user': self.user,
                'key': self.key,
                'account': self.account}

    def get_endpoint(self, session, interface='public', **kwargs):
        """Return an endpoint for the client."""
        if interface is plugin.AUTH_INTERFACE:
            return self.auth_url
        else:
            return self.get_access(session).service_catalog.url_for(
                interface=interface, **kwargs)

    def get_auth_state(self):
        """Retrieve the current authentication state for the plugin.

        :returns: raw python data (which can be JSON serialized) that can be
                  moved into another plugin (of the same type) to have the
                  same authenticated state.
        """
        if self.auth_ref:
            return self.auth_ref.get_state()

    def set_auth_state(self, data):
        """Install existing authentication state for a plugin.

        Take the output of get_auth_state and install that authentication state
        into the current authentication plugin.
        """
        if data:
            self.auth_ref = self.access_class.from_state(data)
        else:
            self.auth_ref = None

    def get_sp_auth_url(self, *args, **kwargs):
        raise NotImplementedError()

    def get_sp_url(self, *args, **kwargs):
        raise NotImplementedError()

    def get_discovery(self, *args, **kwargs):
        raise NotImplementedError()


class PasswordLoader(loading.BaseLoader):
    """Option handling for the ``v1password`` plugin."""
    plugin_class = PasswordPlugin

    def get_options(self):
        """Return the list of parameters associated with the auth plugin.

        This list may be used to generate CLI or config arguments.
        """
        return [
            loading.Opt('auth-url', required=True,
                        help='Authentication URL'),
            # overload project-name as a way to specify an alternate account,
            # since:
            #   - in a world of just users & passwords, this seems the closest
            #     analog to a project, and
            #   - openstackclient will (or used to?) still require that you
            #     provide one anyway
            loading.Opt('project-name', required=False,
                        help='Swift account to use'),
            loading.Opt('username', required=True,
                        deprecated=[loading.Opt('user-name')],
                        help='Username to login with'),
            loading.Opt('password', required=True, secret=True,
                        help='Password to use'),
        ]