Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ec2: add support for creation and importing of real SSH keys (#2108)

Browse Source 
* ec2: add support for creation and importing of real SSH keys

* setup: lock PyYAML version to avoid incompatibilities
This commit is contained in:
Daniel Miranda 2019-05-25 07:17:52 -03:00 committed by Terry Cain
parent d8ff67197b
commit fb2a76fd66
8 changed files with 178 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
moto/ec2/exceptions.py
View File

@ -58,6 +58,14 @@ class InvalidKeyPairDuplicateError(EC2ClientError):
.format(key))
class InvalidKeyPairFormatError(EC2ClientError):
def __init__(self):
super(InvalidKeyPairFormatError, self).__init__(
"InvalidKeyPair.Format",
"Key is not in valid OpenSSH public key format")
class InvalidVPCIdError(EC2ClientError):
def __init__(self, vpc_id):

13
moto/ec2/models.py
View File

@ -20,6 +20,7 @@ from boto.ec2.blockdevicemapping import BlockDeviceMapping, BlockDeviceType
from boto.ec2.spotinstancerequest import SpotInstanceRequest as BotoSpotRequest
from boto.ec2.launchspecification import LaunchSpecification
from moto.compat import OrderedDict
from moto.core import BaseBackend
from moto.core.models import Model, BaseModel
@ -43,6 +44,7 @@ from .exceptions import (
InvalidInstanceIdError,
InvalidInternetGatewayIdError,
InvalidKeyPairDuplicateError,
InvalidKeyPairFormatError,
InvalidKeyPairNameError,
InvalidNetworkAclIdError,
InvalidNetworkAttachmentIdError,
@ -120,6 +122,8 @@ from .utils import (
random_customer_gateway_id,
is_tag_filter,
tag_filter_matches,
rsa_public_key_parse,
rsa_public_key_fingerprint
)
INSTANCE_TYPES = json.load(
@ -910,7 +914,14 @@ class KeyPairBackend(object):
def import_key_pair(self, key_name, public_key_material):
if key_name in self.keypairs:
raise InvalidKeyPairDuplicateError(key_name)
keypair = KeyPair(key_name, **random_key_pair())
try:
rsa_public_key = rsa_public_key_parse(public_key_material)
except ValueError:
raise InvalidKeyPairFormatError()
fingerprint = rsa_public_key_fingerprint(rsa_public_key)
keypair = KeyPair(key_name, material=public_key_material, fingerprint=fingerprint)
self.keypairs[key_name] = keypair
return keypair

60
moto/ec2/utils.py
View File

@ -1,10 +1,19 @@
from __future__ import unicode_literals
import base64
import hashlib
import fnmatch
import random
import re
import six
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
import sshpubkeys.exceptions
from sshpubkeys.keys import SSHKey
EC2_RESOURCE_TO_PREFIX = {
'customer-gateway': 'cgw',
'dhcp-options': 'dopt',
@ -453,23 +462,19 @@ def simple_aws_filter_to_re(filter_string):
def random_key_pair():
def random_hex():
return chr(random.choice(list(range(48, 58)) + list(range(97, 102))))
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend())
private_key_material = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption())
public_key_fingerprint = rsa_public_key_fingerprint(private_key.public_key())
def random_fingerprint():
return ':'.join([random_hex() + random_hex() for i in range(20)])
def random_material():
return ''.join([
chr(random.choice(list(range(65, 91)) + list(range(48, 58)) +
list(range(97, 102))))
for i in range(1000)
])
material = "---- BEGIN RSA PRIVATE KEY ----" + random_material() + \
"-----END RSA PRIVATE KEY-----"
return {
'fingerprint': random_fingerprint(),
'material': material
'fingerprint': public_key_fingerprint,
'material': private_key_material.decode('ascii')
}
@ -535,3 +540,28 @@ def generate_instance_identity_document(instance):
}
return document
def rsa_public_key_parse(key_material):
try:
if not isinstance(key_material, six.binary_type):
key_material = key_material.encode("ascii")
decoded_key = base64.b64decode(key_material).decode("ascii")
public_key = SSHKey(decoded_key)
except (sshpubkeys.exceptions.InvalidKeyException, UnicodeDecodeError):
raise ValueError('bad key')
if not public_key.rsa:
raise ValueError('bad key')
return public_key.rsa
def rsa_public_key_fingerprint(rsa_public_key):
key_data = rsa_public_key.public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo)
fingerprint_hex = hashlib.md5(key_data).hexdigest()
fingerprint = re.sub(r'([a-f0-9]{2})(?!$)', r'\1:', fingerprint_hex)
return fingerprint

3
setup.py
View File

@ -28,7 +28,7 @@ install_requires = [
"xmltodict",
"six>1.9",
"werkzeug",
"PyYAML",
"PyYAML==3.13",
"pytz",
"python-dateutil<3.0.0,>=2.1",
"python-jose<4.0.0",
@ -39,6 +39,7 @@ install_requires = [
"responses>=0.9.0",
"idna<2.9,>=2.5",
"cfn-lint",
"sshpubkeys>=3.1.0,<4.0"
]
extras_require = {

0
tests/test_ec2/__init__.py Normal file
View File

15
tests/test_ec2/helpers.py Normal file
View File

@ -0,0 +1,15 @@
import six
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
def rsa_check_private_key(private_key_material):
assert isinstance(private_key_material, six.string_types)
private_key = serialization.load_pem_private_key(
data=private_key_material.encode('ascii'),
backend=default_backend(),
password=None)
assert isinstance(private_key, rsa.RSAPrivateKey)

105
tests/test_ec2/test_key_pairs.py
View File

@ -4,12 +4,46 @@ import tests.backport_assert_raises
from nose.tools import assert_raises
import boto
import six
import sure # noqa
from boto.exception import EC2ResponseError
from moto import mock_ec2_deprecated
from .helpers import rsa_check_private_key
RSA_PUBLIC_KEY_OPENSSH = b"""\
ssh-rsa \
AAAAB3NzaC1yc2EAAAADAQABAAABAQDusXfgTE4eBP50NglSzCSEGnIL6+cr6m3H\
6cZANOQ+P1o/W4BdtcAL3sor4iGi7SOeJgo\8kweyMQrhrt6HaKGgromRiz37LQx\
4YIAcBi4Zd023mO/V7Rc2Chh18mWgLSmA6ng+j37ip6452zxtv0jHAz9pJolbKBp\
JzbZlPN45ZCTk9ck0fSVHRl6VRSSPQcpqi65XpRf+35zNOCGCc1mAOOTmw59Q2a6\
A3t8mL7r91aM5q6QOQm219lctFM8O7HRJnDgmhGpnjRwE1LyKktWTbgFZ4SNWU2X\
qusUO07jKuSxzPumXBeU+JEtx0J1tqZwJlpGt2R+0qN7nKnPl2+hx \
moto@github.com"""
RSA_PUBLIC_KEY_RFC4716 = b"""\
---- BEGIN SSH2 PUBLIC KEY ----
AAAAB3NzaC1yc2EAAAADAQABAAABAQDusXfgTE4eBP50NglSzCSEGnIL6+cr6m3H6cZANO
Q+P1o/W4BdtcAL3sor4iGi7SOeJgo8kweyMQrhrt6HaKGgromRiz37LQx4YIAcBi4Zd023
mO/V7Rc2Chh18mWgLSmA6ng+j37ip6452zxtv0jHAz9pJolbKBpJzbZlPN45ZCTk9ck0fS
VHRl6VRSSPQcpqi65XpRf+35zNOCGCc1mAOOTmw59Q2a6A3t8mL7r91aM5q6QOQm219lct
FM8O7HRJnDgmhGpnjRwE1LyKktWTbgFZ4SNWU2XqusUO07jKuSxzPumXBeU+JEtx0J1tqZ
wJlpGt2R+0qN7nKnPl2+hx
---- END SSH2 PUBLIC KEY ----
"""
RSA_PUBLIC_KEY_FINGERPRINT = "6a:49:07:1c:7e:bd:d2:bd:96:25:fe:b5:74:83:ae:fd"
DSA_PUBLIC_KEY_OPENSSH = b"""ssh-dss \
AAAAB3NzaC1kc3MAAACBAJ0aXctVwbN6VB81gpo8R7DUk8zXRjZvrkg8Y8vEGt63gklpNJNsLXtEUXkl5D4c0nD2FZO1rJNqFoe\
OQOCoGSfclHvt9w4yPl/lUEtb3Qtj1j80MInETHr19vaSunRk5R+M+8YH+LLcdYdz7MijuGey02mbi0H9K5nUIcuLMArVAAAAFQ\
D0RDvsObRWBlnaW8645obZBM86jwAAAIBNZwf3B4krIzAwVfkMHLDSdAvs7lOWE7o8SJLzr9t4a9HhYp9SLbMzJ815KWfidEYV2\
+s4ZaPCfcZ1GENFRbE8rixz5eMAjEUXEPMJkblDZTHzMsH96z2cOCQZ0vfOmgznsf18Uf725pqo9OqAioEsTJjX8jtI2qNPEBU0\
uhMSZQAAAIBBMGhDu5CWPUlS2QG7vzmzw81XasmHE/s2YPDRbolkriwlunpgwZhCscoQP8HFHY+DLUVvUb+GZwBmFt4l1uHl03b\
ffsm7UIHtCBYERr9Nx0u20ldfhkgB1lhaJb5o0ZJ3pmJ38KChfyHe5EUcqRdEFo89Mp72VI2Z6UHyL175RA== \
moto@github.com"""
@mock_ec2_deprecated
def test_key_pairs_empty():
@ -33,14 +67,15 @@ def test_key_pairs_create():
conn = boto.connect_ec2('the_key', 'the_secret')
with assert_raises(EC2ResponseError) as ex:
kp = conn.create_key_pair('foo', dry_run=True)
conn.create_key_pair('foo', dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the CreateKeyPair operation: Request would have succeeded, but DryRun flag is set')
kp = conn.create_key_pair('foo')
assert kp.material.startswith('---- BEGIN RSA PRIVATE KEY ----')
rsa_check_private_key(kp.material)
kps = conn.get_all_key_pairs()
assert len(kps) == 1
assert kps[0].name == 'foo'
@ -49,13 +84,19 @@ def test_key_pairs_create():
@mock_ec2_deprecated
def test_key_pairs_create_two():
conn = boto.connect_ec2('the_key', 'the_secret')
kp = conn.create_key_pair('foo')
kp = conn.create_key_pair('bar')
assert kp.material.startswith('---- BEGIN RSA PRIVATE KEY ----')
kp1 = conn.create_key_pair('foo')
rsa_check_private_key(kp1.material)
kp2 = conn.create_key_pair('bar')
rsa_check_private_key(kp2.material)
assert kp1.material != kp2.material
kps = conn.get_all_key_pairs()
kps.should.have.length_of(2)
[i.name for i in kps].should.contain('foo')
[i.name for i in kps].should.contain('bar')
assert {i.name for i in kps} == {'foo', 'bar'}
kps = conn.get_all_key_pairs('foo')
kps.should.have.length_of(1)
kps[0].name.should.equal('foo')
@ -64,8 +105,7 @@ def test_key_pairs_create_two():
@mock_ec2_deprecated
def test_key_pairs_create_exist():
conn = boto.connect_ec2('the_key', 'the_secret')
kp = conn.create_key_pair('foo')
assert kp.material.startswith('---- BEGIN RSA PRIVATE KEY ----')
conn.create_key_pair('foo')
assert len(conn.get_all_key_pairs()) == 1
with assert_raises(EC2ResponseError) as cm:
@ -105,23 +145,30 @@ def test_key_pairs_import():
conn = boto.connect_ec2('the_key', 'the_secret')
with assert_raises(EC2ResponseError) as ex:
kp = conn.import_key_pair('foo', b'content', dry_run=True)
conn.import_key_pair('foo', RSA_PUBLIC_KEY_OPENSSH, dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal(
'An error occurred (DryRunOperation) when calling the ImportKeyPair operation: Request would have succeeded, but DryRun flag is set')
kp = conn.import_key_pair('foo', b'content')
assert kp.name == 'foo'
kp1 = conn.import_key_pair('foo', RSA_PUBLIC_KEY_OPENSSH)
assert kp1.name == 'foo'
assert kp1.fingerprint == RSA_PUBLIC_KEY_FINGERPRINT
kp2 = conn.import_key_pair('foo2', RSA_PUBLIC_KEY_RFC4716)
assert kp2.name == 'foo2'
assert kp2.fingerprint == RSA_PUBLIC_KEY_FINGERPRINT
kps = conn.get_all_key_pairs()
assert len(kps) == 1
assert kps[0].name == 'foo'
assert len(kps) == 2
assert kps[0].name == kp1.name
assert kps[1].name == kp2.name
@mock_ec2_deprecated
def test_key_pairs_import_exist():
conn = boto.connect_ec2('the_key', 'the_secret')
kp = conn.import_key_pair('foo', b'content')
kp = conn.import_key_pair('foo', RSA_PUBLIC_KEY_OPENSSH)
assert kp.name == 'foo'
assert len(conn.get_all_key_pairs()) == 1
@ -132,6 +179,32 @@ def test_key_pairs_import_exist():
cm.exception.request_id.should_not.be.none
@mock_ec2_deprecated
def test_key_pairs_invalid():
conn = boto.connect_ec2('the_key', 'the_secret')
with assert_raises(EC2ResponseError) as ex:
conn.import_key_pair('foo', b'')
ex.exception.error_code.should.equal('InvalidKeyPair.Format')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal(
'Key is not in valid OpenSSH public key format')
with assert_raises(EC2ResponseError) as ex:
conn.import_key_pair('foo', b'garbage')
ex.exception.error_code.should.equal('InvalidKeyPair.Format')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal(
'Key is not in valid OpenSSH public key format')
with assert_raises(EC2ResponseError) as ex:
conn.import_key_pair('foo', DSA_PUBLIC_KEY_OPENSSH)
ex.exception.error_code.should.equal('InvalidKeyPair.Format')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal(
'Key is not in valid OpenSSH public key format')
@mock_ec2_deprecated
def test_key_pair_filters():
conn = boto.connect_ec2('the_key', 'the_secret')

10
tests/test_ec2/test_utils.py
View File

@ -1,8 +1,12 @@
from moto.ec2 import utils
from .helpers import rsa_check_private_key
def test_random_key_pair():
key_pair = utils.random_key_pair()
assert len(key_pair['fingerprint']) == 59
assert key_pair['material'].startswith('---- BEGIN RSA PRIVATE KEY ----')
assert key_pair['material'].endswith('-----END RSA PRIVATE KEY-----')
rsa_check_private_key(key_pair['material'])
# AWS uses MD5 fingerprints, which are 47 characters long, *not* SHA1
# fingerprints with 59 characters.
assert len(key_pair['fingerprint']) == 47