path: root/openstack/common/rpc/securemessage.py

# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2013 Red Hat, Inc.
#
#    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 collections
import os
import struct
import time

import requests

from oslo.config import cfg

from openstack.common.crypto import utils as cryptoutils
from openstack.common import jsonutils
from openstack.common import log as logging

secure_message_opts = [
    cfg.BoolOpt('enabled', default=True,
                help='Whether Secure Messaging (Signing) is enabled,'
                     ' defaults to enabled'),
    cfg.BoolOpt('enforced', default=False,
                help='Whether Secure Messaging (Signing) is enforced,'
                     ' defaults to not enforced'),
    cfg.BoolOpt('encrypt', default=False,
                help='Whether Secure Messaging (Encryption) is enabled,'
                     ' defaults to not enabled'),
    cfg.StrOpt('secret_keys_file',
               help='Path to the file containing the keys, takes precedence'
                    ' over secret_key'),
    cfg.MultiStrOpt('secret_key',
                    help='A list of keys: (ex: name:<base64 encoded key>),'
                         ' ignored if secret_keys_file is set'),
    cfg.StrOpt('kds_endpoint',
               help='KDS endpoint (ex: http://kds.example.com:35357/v3)'),
]
secure_message_group = cfg.OptGroup('secure_messages',
                                    title='Secure Messaging options')

LOG = logging.getLogger(__name__)


class SecureMessageException(Exception):
    """Generic Exception for Secure Messages."""

    msg = "An unknown Secure Message related exception occurred."

    def __init__(self, msg=None):
        if msg is None:
            msg = self.msg
        super(SecureMessageException, self).__init__(msg)


class SharedKeyNotFound(SecureMessageException):
    """No shared key was found and no other external authentication mechanism
    is available.
    """

    msg = "Shared Key for [%s] Not Found. (%s)"

    def __init__(self, name, errmsg):
        super(SharedKeyNotFound, self).__init__(self.msg % (name, errmsg))


class InvalidMetadata(SecureMessageException):
    """The metadata is invalid."""

    msg = "Invalid metadata: %s"

    def __init__(self, err):
        super(InvalidMetadata, self).__init__(self.msg % err)


class InvalidSignature(SecureMessageException):
    """Signature validation failed."""

    msg = "Failed to validate signature (source=%s, destination=%s)"

    def __init__(self, src, dst):
        super(InvalidSignature, self).__init__(self.msg % (src, dst))


class UnknownDestinationName(SecureMessageException):
    """The Destination name is unknown to us."""

    msg = "Invalid destination name (%s)"

    def __init__(self, name):
        super(UnknownDestinationName, self).__init__(self.msg % name)


class InvalidEncryptedTicket(SecureMessageException):
    """The Encrypted Ticket could not be successfully handled."""

    msg = "Invalid Ticket (source=%s, destination=%s)"

    def __init__(self, src, dst):
        super(InvalidEncryptedTicket, self).__init__(self.msg % (src, dst))


class InvalidExpiredTicket(SecureMessageException):
    """The ticket received is already expired."""

    msg = "Expired ticket (source=%s, destination=%s)"

    def __init__(self, src, dst):
        super(InvalidExpiredTicket, self).__init__(self.msg % (src, dst))


class CommunicationError(SecureMessageException):
    """The Communication with the KDS failed."""

    msg = "Communication Error (target=%s): %s"

    def __init__(self, target, errmsg):
        super(CommunicationError, self).__init__(self.msg % (target, errmsg))


class InvalidArgument(SecureMessageException):
    """Bad initialization argument."""

    msg = "Invalid argument: %s"

    def __init__(self, errmsg):
        super(InvalidArgument, self).__init__(self.msg % errmsg)


Ticket = collections.namedtuple('Ticket', ['skey', 'ekey', 'esek'])


class KeyStore(object):
    """A storage class for Signing and Encryption Keys.

    This class creates an object that holds Generic Keys like Signing
    Keys, Encryption Keys, Encrypted SEK Tickets ...
    """

    def __init__(self):
        self._kvps = dict()

    def _get_key_name(self, source, target, ktype):
        return (source, target, ktype)

    def _put(self, src, dst, ktype, expiration, data):
        name = self._get_key_name(src, dst, ktype)
        self._kvps[name] = (expiration, data)

    def _get(self, src, dst, ktype):
        name = self._get_key_name(src, dst, ktype)
        if name in self._kvps:
            expiration, data = self._kvps[name]
            if expiration > time.time():
                return data
            else:
                del self._kvps[name]

        return None

    def clear(self):
        """Wipes the store clear of all data."""
        self._kvps.clear()

    def put_ticket(self, source, target, skey, ekey, esek, expiration):
        """Puts a sek pair in the cache.

        :param source: Client name
        :param target: Target name
        :param skey: The Signing Key
        :param ekey: The Encription Key
        :param esek: The token encrypted with the target key
        :param expiration: Expiration time in seconds since Epoch
        """
        keys = Ticket(skey, ekey, esek)
        self._put(source, target, 'ticket', expiration, keys)

    def get_ticket(self, source, target):
        """Returns a Ticket (skey, ekey, esek) namedtuple for the
           source/target pair.
        """
        return self._get(source, target, 'ticket')


_KEY_STORE = KeyStore()


class _KDSClient(object):

    USER_AGENT = 'oslo-incubator/rpc'

    def __init__(self, endpoint=None, timeout=None):
        """A KDS Client class."""

        self._endpoint = endpoint
        if timeout is not None:
            self.timeout = float(timeout)
        else:
            self.timeout = None

    def _do_post(self, url, request):
        req_kwargs = dict()
        req_kwargs['headers'] = dict()
        req_kwargs['headers']['User-Agent'] = self.USER_AGENT
        req_kwargs['headers']['Content-Type'] = 'application/json'
        req_kwargs['data'] = jsonutils.dumps({'request': request})
        if self.timeout is not None:
            req_kwargs['timeout'] = self.timeout

        try:
            resp = requests.post(url, **req_kwargs)
        except requests.ConnectionError as e:
            err = "Unable to establish connection. %s" % e
            raise CommunicationError(url, err)

        return resp

    def _get_reply(self, url, resp):
        if resp.text:
            try:
                body = jsonutils.loads(resp.text)
                reply = body['reply']
            except (KeyError, TypeError, ValueError):
                msg = "Failed to decode reply: %s" % resp.text
                raise CommunicationError(url, msg)
        else:
            msg = "No reply data was returned."
            raise CommunicationError(url, msg)

        return reply

    def _get_ticket(self, request, url=None, redirects=10):
        """Send an HTTP request.

        Wraps around 'requests' to handle redirects and common errors.
        """
        if url is None:
            if not self._endpoint:
                raise CommunicationError(url, 'Endpoint not configured')
            url = self._endpoint + '/kds/ticket/' + request['signature']

        while redirects:
            resp = self._do_post(url, request)
            if resp.status_code in (301, 302, 305):
                # Redirected. Reissue the request to the new location.
                url = resp.headers['location']
                redirects -= 1
                continue
            elif resp.status_code != 200:
                msg = "Request returned failure status: %s (%s)"
                err = msg % (resp.status_code, resp.text)
                raise CommunicationError(url, err)

            return self._get_reply(url, resp)

        raise CommunicationError(url, "Too many redirections, giving up!")

    def get_ticket(self, source, target, crypto, key):

        # prepare metadata
        md = {'requestor': source,
              'target': target,
              'timestamp': time.time(),
              'nonce': struct.unpack('Q', os.urandom(8))[0]}
        metadata = base64.b64encode(jsonutils.dumps(md))

        # sign metadata
        signature = crypto.sign(key, metadata)

        # HTTP request
        reply = self._get_ticket({'metadata': metadata,
                                  'signature': signature})

        # verify reply
        signature = crypto.sign(key, (reply['metadata'] + reply['ticket']))
        if signature != reply['signature']:
            raise InvalidEncryptedTicket(md['source'], md['destination'])
        md = jsonutils.loads(base64.b64decode(reply['metadata']))
        if ((md['source'] != source or
             md['destination'] != target or
             md['expiration'] < time.time())):
            raise InvalidEncryptedTicket(md['source'], md['destination'])

        # return ticket data
        tkt = jsonutils.loads(crypto.decrypt(key, reply['ticket']))

        return tkt, md['expiration']


# we need to keep a global nonce, as this value should never repeat non
# matter how many SecureMessage objects we create
_NONCE = None


def _get_nonce():
    """We keep a single counter per instance, as it is so huge we can't
    possibly cycle through within 1/100 of a second anyway.
    """

    global _NONCE
    # Lazy initialize, for now get a random value, multiply by 2^32 and
    # use it as the nonce base. The counter itself will rotate after
    # 2^32 increments.
    if _NONCE is None:
        _NONCE = [struct.unpack('I', os.urandom(4))[0], 0]

    # Increment counter and wrap at 2^32
    _NONCE[1] += 1
    if _NONCE[1] > 0xffffffff:
        _NONCE[1] = 0

    # Return base + counter
    return long((_NONCE[0] * 0xffffffff)) + _NONCE[1]


class SecureMessage(object):
    """A Secure Message object.

    This class creates a signing/encryption facility for RPC messages.
    It encapsulates all the necessary crypto primitives to insulate
    regular code from the intricacies of message authentication, validation
    and optionally encryption.

    :param topic: The topic name of the queue
    :param host: The server name, together with the topic it forms a unique
                 name that is used to source signing keys, and verify
                 incoming messages.
    :param conf: a ConfigOpts object
    :param key: (optional) explicitly pass in endpoint private key.
                  If not provided it will be sourced from the service config
    :param key_store: (optional) Storage class for local caching
    :param encrypt: (defaults to False) Whether to encrypt messages
    :param enctype: (defaults to AES) Cipher to use
    :param hashtype: (defaults to SHA256) Hash function to use for signatures
    """

    def __init__(self, topic, host, conf, key=None, key_store=None,
                 encrypt=None, enctype='AES', hashtype='SHA256'):

        conf.register_group(secure_message_group)
        conf.register_opts(secure_message_opts, group='secure_messages')

        self._name = '%s.%s' % (topic, host)
        self._key = key
        self._conf = conf.secure_messages
        self._encrypt = self._conf.encrypt if (encrypt is None) else encrypt
        self._crypto = cryptoutils.SymmetricCrypto(enctype, hashtype)
        self._hkdf = cryptoutils.HKDF(hashtype)
        self._kds = _KDSClient(self._conf.kds_endpoint)

        if self._key is None:
            self._key = self._init_key(topic, self._name)
        if self._key is None:
            err = "Secret Key (or key file) is missing or malformed"
            raise SharedKeyNotFound(self._name, err)

        self._key_store = key_store or _KEY_STORE

    def _init_key(self, topic, name):
        keys = None
        if self._conf.secret_keys_file:
            with open(self._conf.secret_keys_file, 'r') as f:
                keys = f.readlines()
        elif self._conf.secret_key:
            keys = self._conf.secret_key

        if keys is None:
            return None

        for k in keys:
            if k[0] == '#':
                continue
            if ':' not in k:
                break
            svc, key = k.split(':', 1)
            if svc == topic or svc == name:
                return base64.b64decode(key)

        return None

    def _split_key(self, key, size):
        sig_key = key[:size]
        enc_key = key[size:]
        return sig_key, enc_key

    def _decode_esek(self, key, source, target, timestamp, esek):
        """This function decrypts the esek buffer passed in and returns a
        KeyStore to be used to check and decrypt the received message.

        :param key: The key to use to decrypt the ticket (esek)
        :param source: The name of the source service
        :param traget: The name of the target service
        :param timestamp: The incoming message timestamp
        :param esek: a base64 encoded encrypted block containing a JSON string
        """
        rkey = None

        try:
            s = self._crypto.decrypt(key, esek)
            j = jsonutils.loads(s)

            rkey = base64.b64decode(j['key'])
            expiration = j['timestamp'] + j['ttl']
            if j['timestamp'] > timestamp or timestamp > expiration:
                raise InvalidExpiredTicket(source, target)

        except Exception:
            raise InvalidEncryptedTicket(source, target)

        info = '%s,%s,%s' % (source, target, str(j['timestamp']))

        sek = self._hkdf.expand(rkey, info, len(key) * 2)

        return self._split_key(sek, len(key))

    def _get_ticket(self, target):
        """This function will check if we already have a SEK for the specified
        target in the cache, or will go and try to fetch a new SEK from the key
        server.

        :param target: The name of the target service
        """
        ticket = self._key_store.get_ticket(self._name, target)

        if ticket is not None:
            return ticket

        tkt, expiration = self._kds.get_ticket(self._name, target,
                                               self._crypto, self._key)

        self._key_store.put_ticket(self._name, target,
                                   base64.b64decode(tkt['skey']),
                                   base64.b64decode(tkt['ekey']),
                                   tkt['esek'], expiration)
        return self._key_store.get_ticket(self._name, target)

    def encode(self, version, target, json_msg):
        """This is the main encoding function.

        It takes a target and a message and returns a tuple consisting of a
        JSON serialized metadata object, a JSON serialized (and optionally
        encrypted) message, and a signature.

        :param version: the current envelope version
        :param target: The name of the target service (usually with hostname)
        :param json_msg: a serialized json message object
        """
        ticket = self._get_ticket(target)

        metadata = jsonutils.dumps({'source': self._name,
                                    'destination': target,
                                    'timestamp': time.time(),
                                    'nonce': _get_nonce(),
                                    'esek': ticket.esek,
                                    'encryption': self._encrypt})

        message = json_msg
        if self._encrypt:
            message = self._crypto.encrypt(ticket.ekey, message)

        signature = self._crypto.sign(ticket.skey,
                                      version + metadata + message)

        return (metadata, message, signature)

    def decode(self, version, metadata, message, signature):
        """This is the main decoding function.

        It takes a version, metadata, message and signature strings and
        returns a tuple with a (decrypted) message and metadata or raises
        an exception in case of error.

        :param version: the current envelope version
        :param metadata: a JSON serialized object with metadata for validation
        :param message: a JSON serialized (base64 encoded encrypted) message
        :param signature: a base64 encoded signature
        """
        md = jsonutils.loads(metadata)

        check_args = ('source', 'destination', 'timestamp',
                      'nonce', 'esek', 'encryption')
        for arg in check_args:
            if arg not in md:
                raise InvalidMetadata('Missing metadata "%s"' % arg)

        if md['destination'] != self._name:
            # TODO(simo) handle group keys by checking target
            raise UnknownDestinationName(md['destination'])

        try:
            skey, ekey = self._decode_esek(self._key,
                                           md['source'], md['destination'],
                                           md['timestamp'], md['esek'])
        except InvalidExpiredTicket:
            raise
        except Exception:
            raise InvalidMetadata('Failed to decode ESEK for %s/%s' % (
                                  md['source'], md['destination']))

        sig = self._crypto.sign(skey, version + metadata + message)

        if sig != signature:
            raise InvalidSignature(md['source'], md['destination'])

        if md['encryption'] is True:
            msg = self._crypto.decrypt(ekey, message)
        else:
            msg = message

        return (md, msg)