<commit_before># Copyright 2011 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.
"""
Tests for Crypto module.
"""

import os
import StringIO

import mock
from mox3 import mox
from oslo_concurrency import processutils
import paramiko

from nova import crypto
from nova import db
from nova import exception
from nova import test
from nova import utils


class X509Test(test.TestCase):
    def test_can_generate_x509(self):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            crypto.ensure_ca_filesystem()
            _key, cert_str = crypto.generate_x509_cert('fake', 'fake')

            project_cert = crypto.fetch_ca(project_id='fake')

            signed_cert_file = os.path.join(tmpdir, "signed")
            with open(signed_cert_file, 'w') as keyfile:
                keyfile.write(cert_str)

            project_cert_file = os.path.join(tmpdir, "project")
            with open(project_cert_file, 'w') as keyfile:
                keyfile.write(project_cert)

            enc, err = utils.execute('openssl', 'verify', '-CAfile',
                    project_cert_file, '-verbose', signed_cert_file)
            self.assertFalse(err)

    def test_encrypt_decrypt_x509(self):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            project_id = "fake"
            crypto.ensure_ca_filesystem()
            cert = crypto.fetch_ca(project_id)
            public_key = os.path.join(tmpdir, "public.pem")
            with open(public_key, 'w') as keyfile:
                keyfile.write(cert)
            text = "some @#!%^* test text"
            enc, _err = utils.execute('openssl',
                                     'rsautl',
                                     '-certin',
                                     '-encrypt',
                                     '-inkey', '%s' % public_key,
                                     process_input=text)
            dec = crypto.decrypt_text(project_id, enc)
            self.assertEqual(text, dec)

    @mock.patch.object(utils, 'execute',
                       side_effect=processutils.ProcessExecutionError)
    def test_ensure_ca_filesystem_chdir(self, *args, **kargs):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            start = os.getcwd()
            self.assertRaises(processutils.ProcessExecutionError,
                              crypto.ensure_ca_filesystem)
            self.assertEqual(start, os.getcwd())


class RevokeCertsTest(test.TestCase):

    def test_revoke_certs_by_user_and_project(self):
        user_id = 'test_user'
        project_id = 2
        file_name = 'test_file'

        def mock_certificate_get_all_by_user_and_project(context,
                                                         user_id,
                                                         project_id):

            return [{"user_id": user_id, "project_id": project_id,
                                          "file_name": file_name}]

        self.stubs.Set(db, 'certificate_get_all_by_user_and_project',
                           mock_certificate_get_all_by_user_and_project)

        self.mox.StubOutWithMock(crypto, 'revoke_cert')
        crypto.revoke_cert(project_id, file_name)

        self.mox.ReplayAll()

        crypto.revoke_certs_by_user_and_project(user_id, project_id)

    def test_revoke_certs_by_user(self):
        user_id = 'test_user'
        project_id = 2
        file_name = 'test_file'

        def mock_certificate_get_all_by_user(context, user_id):

            return [{"user_id": user_id, "project_id": project_id,
                                          "file_name": file_name}]

        self.stubs.Set(db, 'certificate_get_all_by_user',
                                    mock_certificate_get_all_by_user)

        self.mox.StubOutWithMock(crypto, 'revoke_cert')
        crypto.revoke_cert(project_id, mox.IgnoreArg())

        self.mox.ReplayAll()

        crypto.revoke_certs_by_user(user_id)

    def test_revoke_certs_by_project(self):
        user_id = 'test_user'
        project_id = 2
        file_name = 'test_file'

        def mock_certificate_get_all_by_project(context, project_id):

            return [{"user_id": user_id, "project_id": project_id,
                                          "file_name": file_name}]

        self.stubs.Set(db, 'certificate_get_all_by_project',
                                    mock_certificate_get_all_by_project)

        self.mox.StubOutWithMock(crypto, 'revoke_cert')
        crypto.revoke_cert(project_id, mox.IgnoreArg())

        self.mox.ReplayAll()

        crypto.revoke_certs_by_project(project_id)

    @mock.patch.object(utils, 'execute',
                       side_effect=processutils.ProcessExecutionError)
    @mock.patch.object(os, 'chdir', return_value=None)
    def test_revoke_cert_process_execution_error(self, *args, **kargs):
        self.assertRaises(exception.RevokeCertFailure, crypto.revoke_cert,
                          2, 'test_file')

    @mock.patch.object(os, 'chdir', mock.Mock(side_effect=OSError))
    def test_revoke_cert_project_not_found_chdir_fails(self, *args, **kargs):
        self.assertRaises(exception.ProjectNotFound, crypto.revoke_cert,
                          2, 'test_file')


class CertExceptionTests(test.TestCase):
    def test_fetch_ca_file_not_found(self):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            self.flags(use_project_ca=True)

            self.assertRaises(exception.CryptoCAFileNotFound, crypto.fetch_ca,
                              project_id='fake')

    def test_fetch_crl_file_not_found(self):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            self.flags(use_project_ca=True)

            self.assertRaises(exception.CryptoCRLFileNotFound,
                              crypto.fetch_crl, project_id='fake')


class EncryptionTests(test.TestCase):
    pubkey = ("ssh-rsa A<KEY>2EAAAADAQABAAABAQDArtgrfBu/g2o28o+H2ng/crv"
              "<KEY>"
              "<KEY>"
              "<KEY>"
              "<KEY>"
              "<KEY>"
              "uF test@test")
    prikey = """-----BEGIN RSA PRIVATE KEY-----
<KEY>
<KEY>gPhAZPRWJZ+anrEkyR4iLq6ZoPZ3BQn97U=
-----END RSA PRIVATE KEY-----"""
    text = "Some text! %$*"

    def _ssh_decrypt_text(self, ssh_private_key, text):
        with utils.tempdir() as tmpdir:
            sshkey = os.path.abspath(os.path.join(tmpdir, 'ssh.key'))
            with open(sshkey, 'w') as f:
                f.write(ssh_private_key)
            try:
                dec, _err = utils.execute('openssl',
                                          'rsautl',
                                          '-decrypt',
                                          '-inkey', sshkey,
                                          process_input=text)
                return dec
            except processutils.ProcessExecutionError as exc:
                raise exception.DecryptionFailure(reason=exc.stderr)

    def test_ssh_encrypt_decrypt_text(self):
        enc = crypto.ssh_encrypt_text(self.pubkey, self.text)
        self.assertNotEqual(enc, self.text)
        result = self._ssh_decrypt_text(self.prikey, enc)
        self.assertEqual(result, self.text)

    def test_ssh_encrypt_failure(self):
        self.assertRaises(exception.EncryptionFailure,
                          crypto.ssh_encrypt_text, '', self.text)


class ConversionTests(test.TestCase):
    k1 = ("ssh-rsa AAAAB3NzaC1yc2EAAAABIwAAAQEA4CqmrxfU7x4sJrubpMNxeglul+d"
          "ByrsicnvQcHDEjPzdvoz+BaoAG9bjCA5mCeTBIISsVTVXz/hxNeiuBV6LH/UR/c"
          "27yl53ypN+821ImoexQZcKItdnjJ3gVZlDob1f9+1qDVy63NJ1c+TstkrCTRVeo"
          "9VyE7RpdSS4UCiBe8Xwk3RkedioFxePrI0Ktc2uASw2G0G2Rl7RN7KZOJbCivfF"
          "LQMAOu6e+7fYvuE1gxGHHj7dxaBY/ioGOm1W4JmQ1V7AKt19zTBlZKduN8FQMSF"
          "r35CDlvoWs0+OP8nwlebKNCi/5sdL8qiSLrAcPB4LqdkAf/blNSVA2Yl83/c4lQ"
          "== test@test")

    k2 = ("-----BEGIN PUBLIC KEY-----\n"
          "MIIBIDANBgkqhkiG9w0BAQEFAAOCAQ0AMIIBCAKCAQEA4CqmrxfU7x4sJrubpMNx\n"
          "eglul+dByrsicnvQcHDEjPzdvoz+<KEY>n"
          "<KEY>"
          "TRVeo9VyE7RpdSS4UCiBe8Xwk3RkedioFxePrI0Ktc2uASw2G0G2Rl7RN7KZOJbC\n"
          "<KEY>zTBlZKduN8FQ\n"
          "MSFr35CDlvoWs0+OP8nwlebKNCi/5sdL8qiSLrAcPB4LqdkAf/blNSVA2Yl83/c4\n"
          "lQIBIw==\n"
          "-----END PUBLIC KEY-----\n")

    def test_convert_keys(self):
        result = crypto.convert_from_sshrsa_to_pkcs8(self.k1)
        self.assertEqual(result, self.k2)

    def test_convert_failure(self):
        self.assertRaises(exception.EncryptionFailure,
                          crypto.convert_from_sshrsa_to_pkcs8, '')


class KeyPairTest(test.TestCase):
    rsa_prv = (
        "-----BEGIN RSA PRIVATE KEY-----\n"
        "<KEY>n"
        "WcKlSZBaWPbwbt20/r74eMGZPlqtEi9Ro+EHj4/n5+3A2Mh11h0PGSt53PSPfWwo\n"
        "ZhEg9hQ1w1ZxfBMCx7eG2YdGFQocMgR0zQasJGjjt8hruCnWRB3pNH9DhEwKhgET\n"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "/tRY757oRZzNmq+5V3R52V9WC3qeCBmq3EjWdwJDAphd72/YoOmNMKiPsphKntwI\n"
        "JRS5wodNPlSuYSwEMUypM3f7ttAEn5CASgYgribBDapm7EqkVa2AqSvpFzNvN3/e\n"
        "Sc36/XlxJin7AkKVOnRksuVOOj504VUQfXgVWZkfTeZqAROgA1FSnjUAffcubJmq\n"
        "pDL/JSgOqN4S+sJkkTrb19MuM9M/IdXteloynF+GUKZx6FdVQQc8xCiXgeupeeSD\n"
        "fNMAP7DRAoGBAP0JRFm3fCAavBREKVOyZm20DpeR6zMrVP7ht0SykkT/bw/kiRG+\n"
        "FH1tNioj9uyixt5SiKhH3ZVAunjsKvrwET8i3uz1M2Gk+ovWdLXurBogYNNWafjQ\n"
        "hRhFHpyExoZYRsn58bvYvjFXTO6JxuNS2b59DGBRkQ5mpsOhxarfbZnXAoGBAOcb\n"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        <KEY>"
        "<KEY>DFGQssv8sdletWAFE9/dpUk3A1eiFfC1wGCKuZCDBxKPvOJQjO3uryt\n"
        "d1JGxQkLZ0eVGg+E1O10iC8CgYB4w2QRfNPqllu8D6EPkVHJfeonltgmKOTajm+V\n"
        "HO+kw7OKeLP7EkVU3j+kcSZC8LUQRKZWu1qG2Jtu+7zz+OmYObPygXNNpS56rQW1\n"
        "Yixc/FB3knpEN2DvlilAfxAoGYjD/CL4GhCtdAoZZx0Opc262OEpr4v6hzSb7i4K\n"
        "4KUoXQKBgHfbiaSilxx9guUqvSaexpHmtiUwx05a05fD6tu8Cofl6AM9wGpw3xOT\n"
        "tfo4ehvS13tTz2RDE2xKuetMmkya7UgifcxUmBzqkOlgr0oOi2rp+eDKXnzUUqsH\n"
        "V7E96Dj36K8q2+gZIXcNqjN7PzfkF8pA0G+E1veTi8j5dnvIsy1x\n"
        "-----END RSA PRIVATE KEY-----\n"
    )

    rsa_pub = (
        "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDkbjgPqUSAyPpxHAI/J2wyXVVE"
        "<KEY>"
        "<KEY>"
        "yGu4KdZEHek0f0OETAqGARMfT8IXPFKHR5laz87gZJ/i6nCZEb/Vi73S+dVmrcBq"
        "fC+9blQGTqDi5ZvoVMRLzruQDLhvkRwfpZHeUsTT2ia4b82q0f6qSAh1ByicOSaS"
        "4rfF/mCjoZLASh3R6szUbW974LjD8wP3wpnhmWif2HLugTatyodd/nyBm05N Gen"
        "erated-by-Nova"
    )

    rsa_fp = "e7:66:a1:2c:4f:90:6e:11:19:da:ac:c2:69:e1:ad:89"

    dss_pub = (
        "ssh-dss A<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "wUFmFDwBCGTLl8puVZv7VSu1FG8ko+mzqNebqcN4RMC26NxJqe+RRubn5KtmLoIa"
        "<KEY>"
        "RNnRX7zj5w== Generated-by-Nova"
    )

    dss_fp = "b9:dc:ac:57:df:2a:2b:cf:65:a8:c3:4e:9d:4a:82:3c"

    ecdsa_pub = (
        "<KEY>"
        "<KEY>"
        "uYREz7iLRCP7BwUt8R+ZWzFZDeOLIWU= Generated-by-Nova"
    )

    ecdsa_fp = "16:6a:c9:ec:80:4d:17:3e:d5:3b:6f:c0:d7:15:04:40"

    def test_generate_fingerprint(self):
        fingerprint = crypto.generate_fingerprint(self.rsa_pub)
        self.assertEqual(self.rsa_fp, fingerprint)

        fingerprint = crypto.generate_fingerprint(self.dss_pub)
        self.assertEqual(self.dss_fp, fingerprint)

        fingerprint = crypto.generate_fingerprint(self.ecdsa_pub)
        self.assertEqual(self.ecdsa_fp, fingerprint)

    def test_generate_key_pair(self):
        (private_key, public_key, fingerprint) = crypto.generate_key_pair()
        raw_pub = public_key.split(' ')[1].decode('base64')
        pkey = paramiko.rsakey.RSAKey(None, raw_pub)
        self.assertEqual(2048, pkey.get_bits())

        bits = 4096
        (private_key, public_key, fingerprint) = crypto.generate_key_pair(bits)
        raw_pub = public_key.split(' ')[1].decode('base64')
        pkey = paramiko.rsakey.RSAKey(None, raw_pub)
        self.assertEqual(bits, pkey.get_bits())

        keyin = StringIO.StringIO()
        keyin.write(self.rsa_prv)
        keyin.seek(0)
        key = paramiko.RSAKey.from_private_key(keyin)
<commit_msg>Split up and improve speed of keygen tests<commit_after># Copyright 2011 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.
"""
Tests for Crypto module.
"""

import os
import StringIO

import mock
from mox3 import mox
from oslo_concurrency import processutils
import paramiko

from nova import crypto
from nova import db
from nova import exception
from nova import test
from nova import utils


class X509Test(test.TestCase):
    def test_can_generate_x509(self):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            crypto.ensure_ca_filesystem()
            _key, cert_str = crypto.generate_x509_cert('fake', 'fake')

            project_cert = crypto.fetch_ca(project_id='fake')

            signed_cert_file = os.path.join(tmpdir, "signed")
            with open(signed_cert_file, 'w') as keyfile:
                keyfile.write(cert_str)

            project_cert_file = os.path.join(tmpdir, "project")
            with open(project_cert_file, 'w') as keyfile:
                keyfile.write(project_cert)

            enc, err = utils.execute('openssl', 'verify', '-CAfile',
                    project_cert_file, '-verbose', signed_cert_file)
            self.assertFalse(err)

    def test_encrypt_decrypt_x509(self):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            project_id = "fake"
            crypto.ensure_ca_filesystem()
            cert = crypto.fetch_ca(project_id)
            public_key = os.path.join(tmpdir, "public.pem")
            with open(public_key, 'w') as keyfile:
                keyfile.write(cert)
            text = "some @#!%^* test text"
            enc, _err = utils.execute('openssl',
                                     'rsautl',
                                     '-certin',
                                     '-encrypt',
                                     '-inkey', '%s' % public_key,
                                     process_input=text)
            dec = crypto.decrypt_text(project_id, enc)
            self.assertEqual(text, dec)

    @mock.patch.object(utils, 'execute',
                       side_effect=processutils.ProcessExecutionError)
    def test_ensure_ca_filesystem_chdir(self, *args, **kargs):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            start = os.getcwd()
            self.assertRaises(processutils.ProcessExecutionError,
                              crypto.ensure_ca_filesystem)
            self.assertEqual(start, os.getcwd())


class RevokeCertsTest(test.TestCase):

    def test_revoke_certs_by_user_and_project(self):
        user_id = 'test_user'
        project_id = 2
        file_name = 'test_file'

        def mock_certificate_get_all_by_user_and_project(context,
                                                         user_id,
                                                         project_id):

            return [{"user_id": user_id, "project_id": project_id,
                                          "file_name": file_name}]

        self.stubs.Set(db, 'certificate_get_all_by_user_and_project',
                           mock_certificate_get_all_by_user_and_project)

        self.mox.StubOutWithMock(crypto, 'revoke_cert')
        crypto.revoke_cert(project_id, file_name)

        self.mox.ReplayAll()

        crypto.revoke_certs_by_user_and_project(user_id, project_id)

    def test_revoke_certs_by_user(self):
        user_id = 'test_user'
        project_id = 2
        file_name = 'test_file'

        def mock_certificate_get_all_by_user(context, user_id):

            return [{"user_id": user_id, "project_id": project_id,
                                          "file_name": file_name}]

        self.stubs.Set(db, 'certificate_get_all_by_user',
                                    mock_certificate_get_all_by_user)

        self.mox.StubOutWithMock(crypto, 'revoke_cert')
        crypto.revoke_cert(project_id, mox.IgnoreArg())

        self.mox.ReplayAll()

        crypto.revoke_certs_by_user(user_id)

    def test_revoke_certs_by_project(self):
        user_id = 'test_user'
        project_id = 2
        file_name = 'test_file'

        def mock_certificate_get_all_by_project(context, project_id):

            return [{"user_id": user_id, "project_id": project_id,
                                          "file_name": file_name}]

        self.stubs.Set(db, 'certificate_get_all_by_project',
                                    mock_certificate_get_all_by_project)

        self.mox.StubOutWithMock(crypto, 'revoke_cert')
        crypto.revoke_cert(project_id, mox.IgnoreArg())

        self.mox.ReplayAll()

        crypto.revoke_certs_by_project(project_id)

    @mock.patch.object(utils, 'execute',
                       side_effect=processutils.ProcessExecutionError)
    @mock.patch.object(os, 'chdir', return_value=None)
    def test_revoke_cert_process_execution_error(self, *args, **kargs):
        self.assertRaises(exception.RevokeCertFailure, crypto.revoke_cert,
                          2, 'test_file')

    @mock.patch.object(os, 'chdir', mock.Mock(side_effect=OSError))
    def test_revoke_cert_project_not_found_chdir_fails(self, *args, **kargs):
        self.assertRaises(exception.ProjectNotFound, crypto.revoke_cert,
                          2, 'test_file')


class CertExceptionTests(test.TestCase):
    def test_fetch_ca_file_not_found(self):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            self.flags(use_project_ca=True)

            self.assertRaises(exception.CryptoCAFileNotFound, crypto.fetch_ca,
                              project_id='fake')

    def test_fetch_crl_file_not_found(self):
        with utils.tempdir() as tmpdir:
            self.flags(ca_path=tmpdir)
            self.flags(use_project_ca=True)

            self.assertRaises(exception.CryptoCRLFileNotFound,
                              crypto.fetch_crl, project_id='fake')


class EncryptionTests(test.TestCase):
    pubkey = ("ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDArtgrfBu/g2o28o+H2ng/crv"
              "zgES91i/NNPPFTOutXelrJ9QiPTPTm+B8yspLsXifmbsmXztNOlBQgQXs6usxb4"
              "fnJKNUZ84Vkp5esbqK/L7eyRqwPvqo7btKBMoAMVX/kUyojMpxb7Ssh6M6Y8cpi"
              "goi+MSDPD7+5yRJ9z4mH9h7MCY6Ejv8KTcNYmVHvRhsFUcVhWcIISlNWUGiG7rf"
              "oki060F5myQN3AXcL8gHG5/Qb1RVkQFUKZ5geQ39/wSyYA1Q65QTba/5G2QNbl2"
              "0eAIBTyKZhN6g88ak+yARa6BLLDkrlP7L4WctHQMLsuXHohQsUO9AcOlVMARgrg"
              "uF test@test")
    prikey = """-----BEGIN RSA PRIVATE KEY-----
MIIEpQIBAAKCAQEAwK7YK3wbv4NqNvKPh9p4P3K<KEY>
-----END RSA PRIVATE KEY-----"""
    text = "Some text! %$*"

    def _ssh_decrypt_text(self, ssh_private_key, text):
        with utils.tempdir() as tmpdir:
            sshkey = os.path.abspath(os.path.join(tmpdir, 'ssh.key'))
            with open(sshkey, 'w') as f:
                f.write(ssh_private_key)
            try:
                dec, _err = utils.execute('openssl',
                                          'rsautl',
                                          '-decrypt',
                                          '-inkey', sshkey,
                                          process_input=text)
                return dec
            except processutils.ProcessExecutionError as exc:
                raise exception.DecryptionFailure(reason=exc.stderr)

    def test_ssh_encrypt_decrypt_text(self):
        enc = crypto.ssh_encrypt_text(self.pubkey, self.text)
        self.assertNotEqual(enc, self.text)
        result = self._ssh_decrypt_text(self.prikey, enc)
        self.assertEqual(result, self.text)

    def test_ssh_encrypt_failure(self):
        self.assertRaises(exception.EncryptionFailure,
                          crypto.ssh_encrypt_text, '', self.text)


class ConversionTests(test.TestCase):
    k1 = ("ssh-rsa AAAAB3NzaC1yc2EAAAABIwAAAQEA4CqmrxfU7x4sJrubpMNxeglul+d"
          "ByrsicnvQcHDEjPzdvoz+BaoAG9bjCA5mCeTBIISsVTVXz/hxNeiuBV6LH/UR/c"
          "27yl53ypN+821ImoexQZcKItdnjJ3gVZlDob1f9+1qDVy63NJ1c+TstkrCTRVeo"
          "9VyE7RpdSS4UCiBe8Xwk3RkedioFxePrI0Ktc2uASw2G0G2Rl7RN7KZOJbCivfF"
          "LQMAOu6e+7fYvuE1gxGHHj7dxaBY/ioGOm1W4JmQ1V7AKt19zTBlZKduN8FQMSF"
          "r35CDlvoWs0+OP8nwlebKNCi/5sdL8qiSLrAcPB4LqdkAf/blNSVA2Yl83/c4lQ"
          "== test@test")

    k2 = ("-----BEGIN PUBLIC KEY-----\n"
          "<KEY>"
          "<KEY>"
          "<KEY>"
          "TRVeo9VyE7RpdSS4UCiBe8Xwk3RkedioFxePrI0Ktc2uASw2G0G2Rl7RN7KZOJbC\n"
          "<KEY>e+7fYvuE1gxGHHj7dxaBY/ioGOm1W4JmQ1V7AKt19zTBlZKduN8FQ\n"
          "MSFr35CDlvoWs0+OP8nwlebKNCi/5sdL8qiSLrAcPB4LqdkAf/blNSVA2Yl83/c4\n"
          "lQIBIw==\n"
          "-----END PUBLIC KEY-----\n")

    def test_convert_keys(self):
        result = crypto.convert_from_sshrsa_to_pkcs8(self.k1)
        self.assertEqual(result, self.k2)

    def test_convert_failure(self):
        self.assertRaises(exception.EncryptionFailure,
                          crypto.convert_from_sshrsa_to_pkcs8, '')


class KeyPairTest(test.TestCase):
    rsa_prv = (
        "-----BEGIN RSA PRIVATE KEY-----\n"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        <KEY>"
        "JRS5wodNPlSuYSwEMUypM3f7ttAEn5CASgYgribBDapm7EqkVa2AqSvpFzNvN3/e\n"
        "Sc36/XlxJin7AkKVOnRksuVOOj504VUQfXgVWZkfTeZqAROgA1FSnjUAffcubJmq\n"
        "pDL/JSgOqN4S+sJkkTrb19MuM9M/IdXteloynF+GUKZx6FdVQQc8xCiXgeupeeSD\n"
        "fNMAP7DRAoGBAP0JRFm3fCAavBREKVOyZm20DpeR6zMrVP7ht0SykkT/bw/kiRG+\n"
        "FH1tNioj9uyixt5SiKhH3ZVAunjsKvrwET8i3uz1M2Gk+ovWdLXurBogYNNWafjQ\n"
        "hRhFHpyExoZYRsn58bvYvjFXTO6JxuNS2b59DGBRkQ5mpsOhxarfbZnXAoGBAOcb\n"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "+4GRFMYVbEwaaJd4jzASJn12M9GuquBBXFMF15DxXFL6lmUXEZYdf83YCRqTY6hi\n"
        "NLgIs+XuxDFGQssv8sdletWAFE9/dpUk3A1eiFfC1wGCKuZCDBxKPvOJQjO3uryt\n"
        "d1JGxQkLZ0eVGg+E1O10iC8CgYB4w2QRfNPqllu8D6EPkVHJfeonltgmKOTajm+V\n"
        "HO+kw7OKeLP7EkVU3j+kcSZC8LUQRKZWu1qG2Jtu+7zz+OmYObPygXNNpS56rQW1\n"
        "Yixc/FB3knpEN2DvlilAfxAoGYjD/CL4GhCtdAoZZx0Opc262OEpr4v6hzSb7i4K\n"
        "4KUoXQKBgHfbiaSilxx9guUqvSaexpHmtiUwx05a05fD6tu8Cofl6AM9wGpw3xOT\n"
        "tfo4ehvS13tTz2RDE2xKuetMmkya7UgifcxUmBzqkOlgr0oOi2rp+eDKXnzUUqsH\n"
        "V7E96Dj36K8q2+gZIXcNqjN7PzfkF8pA0G+E1veTi8j5dnvIsy1x\n"
        "-----END RSA PRIVATE KEY-----\n"
    )

    rsa_pub = (
        "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDkbjgPqUSAyPpxHAI/J2wyXVVE"
        "3YH0NXJXmWbCywZ4kKXLBkxZwqVJkFpY9vBu3bT+vvh4wZk+Wq0SL1Gj4QePj+fn"
        "7cDYyHXWHQ8ZK3nc9I99bChmESD2FDXDVnF8EwLHt4bZh0YVChwyBHTNBqwkaOO3"
        "yGu4KdZEHek0f0OETAqGARMfT8IXPFKHR5laz87gZJ/i6nCZEb/Vi73S+dVmrcBq"
        "fC+9blQGTqDi5ZvoVMRLzruQDLhvkRwfpZHeUsTT2ia4b82q0f6qSAh1ByicOSaS"
        "4rfF/mCjoZLASh3R6szUbW974LjD8wP3wpnhmWif2HLugTatyodd/nyBm05N Gen"
        "erated-by-Nova"
    )

    rsa_fp = "e7:66:a1:2c:4f:90:6e:11:19:da:ac:c2:69:e1:ad:89"

    dss_pub = (
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>"
        "<KEY>WV70"
        "RNnRX7zj5w== Generated-by-Nova"
    )

    dss_fp = "b9:dc:ac:57:df:2a:2b:cf:65:a8:c3:4e:9d:4a:82:3c"

    ecdsa_pub = (
        "<KEY>"
        "<KEY>"
        "uYREz7iLRCP7BwUt8R+ZWzFZDeOLIWU= Generated-by-Nova"
    )

    ecdsa_fp = "16:6a:c9:ec:80:4d:17:3e:d5:3b:6f:c0:d7:15:04:40"

    def test_generate_fingerprint(self):
        fingerprint = crypto.generate_fingerprint(self.rsa_pub)
        self.assertEqual(self.rsa_fp, fingerprint)

        fingerprint = crypto.generate_fingerprint(self.dss_pub)
        self.assertEqual(self.dss_fp, fingerprint)

        fingerprint = crypto.generate_fingerprint(self.ecdsa_pub)
        self.assertEqual(self.ecdsa_fp, fingerprint)

    def test_generate_key_pair_2048_bits(self):
        (private_key, public_key, fingerprint) = crypto.generate_key_pair()
        raw_pub = public_key.split(' ')[1].decode('base64')
        pkey = paramiko.rsakey.RSAKey(None, raw_pub)
        self.assertEqual(2048, pkey.get_bits())

    def test_generate_key_pair_1024_bits(self):
        bits = 1024
        (private_key, public_key, fingerprint) = crypto.generate_key_pair(bits)
        raw_pub = public_key.split(' ')[1].decode('base64')
        pkey = paramiko.rsakey.RSAKey(None, raw_pub)
        self.assertEqual(bits, pkey.get_bits())

    def test_generate_key_pair_mocked_private_key(self):
        keyin = StringIO.StringIO()
        keyin.write(self.rsa_prv)
        keyin.seek(0)
        key = paramiko.RSAKey.from_private_key(keyin)

{'fraction_non_alphanumeric': 0.09916528830249645, 'fraction_numerical': 0.03234988652536831, 'mean_word_length': 2.3944379161770466, 'pattern_counts': {'":': 18, '<': 63, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 50, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3078372', 'n_tokens_mistral': 10427, 'n_tokens_neox': 9883, 'n_words': 1191}