path: root/nova/crypto.py

# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# 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.

"""Wrappers around standard crypto data elements.

Includes root and intermediate CAs, SSH key_pairs and x509 certificates.
"""

import base64
import binascii
import io
import os
import typing as ty

from castellan.common import exception as castellan_exception
from castellan.common.objects import passphrase
from castellan import key_manager
from cryptography.hazmat import backends
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography import x509
from oslo_concurrency import processutils
from oslo_log import log as logging
from oslo_utils.secretutils import md5
import paramiko

import nova.conf
from nova import context as nova_context
from nova import exception
from nova.i18n import _
from nova import objects
from nova import utils


LOG = logging.getLogger(__name__)

CONF = nova.conf.CONF

_KEYMGR = None

_VTPM_SECRET_BYTE_LENGTH = 384


def _get_key_manager():
    global _KEYMGR
    if _KEYMGR is None:
        _KEYMGR = key_manager.API(configuration=CONF)
    return _KEYMGR


def generate_fingerprint(public_key: str) -> str:
    try:
        pub_bytes = public_key.encode('utf-8')
        # Test that the given public_key string is a proper ssh key. The
        # returned object is unused since pyca/cryptography does not have a
        # fingerprint method.
        serialization.load_ssh_public_key(
            pub_bytes, backends.default_backend())
        pub_data = base64.b64decode(public_key.split(' ')[1])
        raw_fp = md5(pub_data, usedforsecurity=False).hexdigest()
        return ':'.join(a + b for a, b in zip(raw_fp[::2], raw_fp[1::2]))
    except Exception:
        raise exception.InvalidKeypair(
            reason=_('failed to generate fingerprint'))


def generate_x509_fingerprint(pem_key: ty.Union[bytes, str]) -> str:
    try:
        if isinstance(pem_key, str):
            pem_key = pem_key.encode('utf-8')
        cert = x509.load_pem_x509_certificate(
            pem_key, backends.default_backend())
        raw_fp = binascii.hexlify(
            cert.fingerprint(hashes.SHA1())
        ).decode('ascii')
        return ':'.join(a + b for a, b in zip(raw_fp[::2], raw_fp[1::2]))
    except (ValueError, TypeError, binascii.Error) as ex:
        raise exception.InvalidKeypair(
            reason=_('failed to generate X509 fingerprint. '
                     'Error message: %s') % ex)


def generate_key_pair(bits: int = 2048) -> ty.Tuple[str, str, str]:
    key = paramiko.RSAKey.generate(bits)
    keyout = io.StringIO()
    key.write_private_key(keyout)
    private_key = keyout.getvalue()
    public_key = '%s %s Generated-by-Nova' % (key.get_name(), key.get_base64())
    fingerprint = generate_fingerprint(public_key)
    return (private_key, public_key, fingerprint)


def ssh_encrypt_text(ssh_public_key: str, text: ty.Union[str, bytes]) -> bytes:
    """Encrypt text with an ssh public key.

    If text is a Unicode string, encode it to UTF-8.
    """
    if isinstance(text, str):
        text = text.encode('utf-8')
    try:
        pub_bytes = ssh_public_key.encode('utf-8')
        pub_key = serialization.load_ssh_public_key(
            pub_bytes, backends.default_backend())
        return pub_key.encrypt(text, padding.PKCS1v15())
    except Exception as exc:
        raise exception.EncryptionFailure(reason=str(exc))


def generate_winrm_x509_cert(
    user_id: str,
    bits: int = 2048
) -> ty.Tuple[str, str, str]:
    """Generate a cert for passwordless auth for user in project."""
    subject = '/CN=%s' % user_id
    upn = '%s@localhost' % user_id

    with utils.tempdir() as tmpdir:
        keyfile = os.path.abspath(os.path.join(tmpdir, 'temp.key'))
        conffile = os.path.abspath(os.path.join(tmpdir, 'temp.conf'))

        _create_x509_openssl_config(conffile, upn)

        out, _ = processutils.execute(
            'openssl', 'req', '-x509', '-nodes', '-days', '3650',
            '-config', conffile, '-newkey', 'rsa:%s' % bits,
            '-outform', 'PEM', '-keyout', keyfile, '-subj', subject,
            '-extensions', 'v3_req_client',
            binary=True)

        certificate = out.decode('utf-8')

        out, _ = processutils.execute(
            'openssl', 'pkcs12', '-export', '-inkey', keyfile, '-password',
            'pass:', process_input=out, binary=True)

        private_key = base64.b64encode(out).decode('ascii')
        fingerprint = generate_x509_fingerprint(certificate)

    return (private_key, certificate, fingerprint)


def _create_x509_openssl_config(conffile: str, upn: str):
    content = ("distinguished_name  = req_distinguished_name\n"
               "[req_distinguished_name]\n"
               "[v3_req_client]\n"
               "extendedKeyUsage = clientAuth\n"
               "subjectAltName = otherName:""1.3.6.1.4.1.311.20.2.3;UTF8:%s\n")

    with open(conffile, 'w') as file:
        file.write(content % upn)


def ensure_vtpm_secret(
    context: nova_context.RequestContext,
    instance: 'objects.Instance',
) -> ty.Tuple[str, str]:
    """Communicates with the key manager service to retrieve or create a secret
    for an instance's emulated TPM.

    When creating a secret, its UUID is saved to the instance's system_metadata
    as ``vtpm_secret_uuid``.

    :param context: Nova auth context.
    :param instance: Instance object.
    :return: A tuple comprising (secret_uuid, passphrase).
    :raise: castellan_exception.ManagedObjectNotFoundError if communication
        with the key manager API fails, or if a vtpm_secret_uuid was present in
        the instance's system metadata but could not be found in the key
        manager service.
    """
    key_mgr = _get_key_manager()

    secret_uuid = instance.system_metadata.get('vtpm_secret_uuid')
    if secret_uuid is not None:
        # Try to retrieve the secret from the key manager
        try:
            secret = key_mgr.get(context, secret_uuid)
            # assert secret_uuid == secret.id ?
            LOG.debug(
                "Found existing vTPM secret with UUID %s.",
                secret_uuid, instance=instance)
            return secret.id, secret.get_encoded()
        except castellan_exception.ManagedObjectNotFoundError:
            LOG.warning(
                "Despite being set on the instance, failed to find a vTPM "
                "secret with UUID %s. This should only happen if the secret "
                "was manually deleted from the key manager service. Your vTPM "
                "is likely to be unrecoverable.",
                secret_uuid, instance=instance)
            raise

    # If we get here, the instance has no vtpm_secret_uuid. Create a new one
    # and register it with the key manager.
    secret = base64.b64encode(os.urandom(_VTPM_SECRET_BYTE_LENGTH))
    # Castellan ManagedObject
    cmo = passphrase.Passphrase(
        secret, name="vTPM secret for instance %s" % instance.uuid)
    secret_uuid = key_mgr.store(context, cmo)
    LOG.debug("Created vTPM secret with UUID %s",
              secret_uuid, instance=instance)

    instance.system_metadata['vtpm_secret_uuid'] = secret_uuid
    instance.save()
    return secret_uuid, secret


def delete_vtpm_secret(
    context: nova_context.RequestContext,
    instance: 'objects.Instance',
):
    """Communicates with the key manager service to destroy the secret for an
    instance's emulated TPM.

    This operation is idempotent: if the instance never had a vTPM secret, OR
    if the secret has already been deleted, it is a no-op.

    The ``vtpm_secret_uuid`` member of the instance's system_metadata is
    cleared as a side effect of this method.

    :param context: Nova auth context.
    :param instance: Instance object.
    :return: None
    :raise: castellan_exception.ManagedObjectNotFoundError if communication
        with the key manager API.
    """
    secret_uuid = instance.system_metadata.get('vtpm_secret_uuid')
    if not secret_uuid:
        return

    key_mgr = _get_key_manager()
    try:
        key_mgr.delete(context, secret_uuid)
        LOG.debug("Deleted vTPM secret with UUID %s",
                  secret_uuid, instance=instance)
    except castellan_exception.ManagedObjectNotFoundError:
        LOG.debug("vTPM secret with UUID %s already deleted or never existed.",
                  secret_uuid, instance=instance)

    del instance.system_metadata['vtpm_secret_uuid']
    instance.save()