Merge pull request #2501 from edekadigital/add-iam-virtual-mfa-device

Add iam virtual mfa device
This commit is contained in:
Jack Danger 2019-10-22 16:01:31 -07:00 committed by GitHub
commit a5d4a68922
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 448 additions and 7 deletions

View File

@ -3244,7 +3244,7 @@
- [ ] describe_events
## iam
58% implemented
61% implemented
- [ ] add_client_id_to_open_id_connect_provider
- [X] add_role_to_instance_profile
- [X] add_user_to_group
@ -3265,7 +3265,7 @@
- [ ] create_service_linked_role
- [ ] create_service_specific_credential
- [X] create_user
- [ ] create_virtual_mfa_device
- [X] create_virtual_mfa_device
- [X] deactivate_mfa_device
- [X] delete_access_key
- [X] delete_account_alias
@ -3289,7 +3289,7 @@
- [X] delete_user
- [ ] delete_user_permissions_boundary
- [X] delete_user_policy
- [ ] delete_virtual_mfa_device
- [X] delete_virtual_mfa_device
- [X] detach_group_policy
- [X] detach_role_policy
- [X] detach_user_policy
@ -3349,7 +3349,7 @@
- [X] list_user_policies
- [ ] list_user_tags
- [X] list_users
- [ ] list_virtual_mfa_devices
- [X] list_virtual_mfa_devices
- [X] put_group_policy
- [ ] put_role_permissions_boundary
- [X] put_role_policy

View File

@ -98,9 +98,9 @@ class TooManyTags(RESTError):
class EntityAlreadyExists(RESTError):
code = 409
def __init__(self):
def __init__(self, message):
super(EntityAlreadyExists, self).__init__(
'EntityAlreadyExists', "Unknown")
'EntityAlreadyExists', message)
class ValidationError(RESTError):

View File

@ -1,5 +1,8 @@
from __future__ import unicode_literals
import base64
import os
import random
import string
import sys
from datetime import datetime
import json
@ -40,6 +43,23 @@ class MFADevice(object):
return iso_8601_datetime_without_milliseconds(self.enable_date)
class VirtualMfaDevice(object):
def __init__(self, device_name):
self.serial_number = 'arn:aws:iam::{0}:mfa{1}'.format(ACCOUNT_ID, device_name)
random_base32_string = ''.join(random.choice(string.ascii_uppercase + '234567') for _ in range(64))
self.base32_string_seed = base64.b64encode(random_base32_string.encode('ascii')).decode('ascii')
self.qr_code_png = base64.b64encode(os.urandom(64)) # this would be a generated PNG
self.enable_date = None
self.user_attribute = None
self.user = None
@property
def enabled_iso_8601(self):
return iso_8601_datetime_without_milliseconds(self.enable_date)
class Policy(BaseModel):
is_attachable = False
@ -596,6 +616,7 @@ class IAMBackend(BaseBackend):
self.open_id_providers = {}
self.policy_arn_regex = re.compile(
r'^arn:aws:iam::[0-9]*:policy/.*$')
self.virtual_mfa_devices = {}
super(IAMBackend, self).__init__()
def _init_managed_policies(self):
@ -1244,6 +1265,21 @@ class IAMBackend(BaseBackend):
"Device {0} already exists".format(serial_number)
)
device = self.virtual_mfa_devices.get(serial_number, None)
if device:
device.enable_date = datetime.utcnow()
device.user = user
device.user_attribute = {
'Path': user.path,
'UserName': user.name,
'UserId': user.id,
'Arn': user.arn,
'CreateDate': user.created_iso_8601,
'PasswordLastUsed': None, # not supported
'PermissionsBoundary': {}, # ToDo: add put_user_permissions_boundary() functionality
'Tags': {} # ToDo: add tag_user() functionality
}
user.enable_mfa_device(
serial_number,
authentication_code_1,
@ -1258,12 +1294,74 @@ class IAMBackend(BaseBackend):
"Device {0} not found".format(serial_number)
)
device = self.virtual_mfa_devices.get(serial_number, None)
if device:
device.enable_date = None
device.user = None
device.user_attribute = None
user.deactivate_mfa_device(serial_number)
def list_mfa_devices(self, user_name):
user = self.get_user(user_name)
return user.mfa_devices.values()
def create_virtual_mfa_device(self, device_name, path):
if not path:
path = '/'
if not path.startswith('/') and not path.endswith('/'):
raise ValidationError('The specified value for path is invalid. '
'It must begin and end with / and contain only alphanumeric characters and/or / characters.')
if any(not len(part) for part in path.split('/')[1:-1]):
raise ValidationError('The specified value for path is invalid. '
'It must begin and end with / and contain only alphanumeric characters and/or / characters.')
if len(path) > 512:
raise ValidationError('1 validation error detected: '
'Value "{}" at "path" failed to satisfy constraint: '
'Member must have length less than or equal to 512')
device = VirtualMfaDevice(path + device_name)
if device.serial_number in self.virtual_mfa_devices:
raise EntityAlreadyExists('MFADevice entity at the same path and name already exists.')
self.virtual_mfa_devices[device.serial_number] = device
return device
def delete_virtual_mfa_device(self, serial_number):
device = self.virtual_mfa_devices.pop(serial_number, None)
if not device:
raise IAMNotFoundException('VirtualMFADevice with serial number {0} doesn\'t exist.'.format(serial_number))
def list_virtual_mfa_devices(self, assignment_status, marker, max_items):
devices = list(self.virtual_mfa_devices.values())
if assignment_status == 'Assigned':
devices = [device for device in devices if device.enable_date]
if assignment_status == 'Unassigned':
devices = [device for device in devices if not device.enable_date]
sorted(devices, key=lambda device: device.serial_number)
max_items = int(max_items)
start_idx = int(marker) if marker else 0
if start_idx > len(devices):
raise ValidationError('Invalid Marker.')
devices = devices[start_idx:start_idx + max_items]
if len(devices) < max_items:
marker = None
else:
marker = str(start_idx + max_items)
return devices, marker
def delete_user(self, user_name):
user = self.get_user(user_name)
if user.managed_policies:
@ -1369,7 +1467,7 @@ class IAMBackend(BaseBackend):
open_id_provider = OpenIDConnectProvider(url, thumbprint_list, client_id_list)
if open_id_provider.arn in self.open_id_providers:
raise EntityAlreadyExists
raise EntityAlreadyExists('Unknown')
self.open_id_providers[open_id_provider.arn] = open_id_provider
return open_id_provider

View File

@ -598,6 +598,33 @@ class IamResponse(BaseResponse):
template = self.response_template(LIST_MFA_DEVICES_TEMPLATE)
return template.render(user_name=user_name, devices=devices)
def create_virtual_mfa_device(self):
path = self._get_param('Path')
virtual_mfa_device_name = self._get_param('VirtualMFADeviceName')
virtual_mfa_device = iam_backend.create_virtual_mfa_device(virtual_mfa_device_name, path)
template = self.response_template(CREATE_VIRTUAL_MFA_DEVICE_TEMPLATE)
return template.render(device=virtual_mfa_device)
def delete_virtual_mfa_device(self):
serial_number = self._get_param('SerialNumber')
iam_backend.delete_virtual_mfa_device(serial_number)
template = self.response_template(DELETE_VIRTUAL_MFA_DEVICE_TEMPLATE)
return template.render()
def list_virtual_mfa_devices(self):
assignment_status = self._get_param('AssignmentStatus', 'Any')
marker = self._get_param('Marker')
max_items = self._get_param('MaxItems', 100)
devices, marker = iam_backend.list_virtual_mfa_devices(assignment_status, marker, max_items)
template = self.response_template(LIST_VIRTUAL_MFA_DEVICES_TEMPLATE)
return template.render(devices=devices, marker=marker)
def delete_user(self):
user_name = self._get_param('UserName')
iam_backend.delete_user(user_name)
@ -1600,6 +1627,7 @@ CREDENTIAL_REPORT_GENERATING = """
</ResponseMetadata>
</GenerateCredentialReportResponse>"""
CREDENTIAL_REPORT_GENERATED = """<GenerateCredentialReportResponse>
<GenerateCredentialReportResult>
<State>COMPLETE</State>
@ -1609,6 +1637,7 @@ CREDENTIAL_REPORT_GENERATED = """<GenerateCredentialReportResponse>
</ResponseMetadata>
</GenerateCredentialReportResponse>"""
CREDENTIAL_REPORT = """<GetCredentialReportResponse>
<GetCredentialReportResult>
<Content>{{ report }}</Content>
@ -1620,6 +1649,7 @@ CREDENTIAL_REPORT = """<GetCredentialReportResponse>
</ResponseMetadata>
</GetCredentialReportResponse>"""
LIST_INSTANCE_PROFILES_FOR_ROLE_TEMPLATE = """<ListInstanceProfilesForRoleResponse>
<ListInstanceProfilesForRoleResult>
<IsTruncated>false</IsTruncated>
@ -1652,6 +1682,7 @@ LIST_INSTANCE_PROFILES_FOR_ROLE_TEMPLATE = """<ListInstanceProfilesForRoleRespon
</ResponseMetadata>
</ListInstanceProfilesForRoleResponse>"""
LIST_MFA_DEVICES_TEMPLATE = """<ListMFADevicesResponse>
<ListMFADevicesResult>
<MFADevices>
@ -1670,6 +1701,61 @@ LIST_MFA_DEVICES_TEMPLATE = """<ListMFADevicesResponse>
</ListMFADevicesResponse>"""
CREATE_VIRTUAL_MFA_DEVICE_TEMPLATE = """<CreateVirtualMFADeviceResponse xmlns="https://iam.amazonaws.com/doc/2010-05-08/">
<CreateVirtualMFADeviceResult>
<VirtualMFADevice>
<SerialNumber>{{ device.serial_number }}</SerialNumber>
<Base32StringSeed>{{ device.base32_string_seed }}</Base32StringSeed>
<QRCodePNG>{{ device.qr_code_png }}</QRCodePNG>
</VirtualMFADevice>
</CreateVirtualMFADeviceResult>
<ResponseMetadata>
<RequestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</RequestId>
</ResponseMetadata>
</CreateVirtualMFADeviceResponse>"""
DELETE_VIRTUAL_MFA_DEVICE_TEMPLATE = """<DeleteVirtualMFADeviceResponse xmlns="https://iam.amazonaws.com/doc/2010-05-08/">
<ResponseMetadata>
<RequestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</RequestId>
</ResponseMetadata>
</DeleteVirtualMFADeviceResponse>"""
LIST_VIRTUAL_MFA_DEVICES_TEMPLATE = """<ListVirtualMFADevicesResponse xmlns="https://iam.amazonaws.com/doc/2010-05-08/">
<ListVirtualMFADevicesResult>
{% if marker is none %}
<IsTruncated>false</IsTruncated>
{% else %}
<IsTruncated>true</IsTruncated>
<Marker>{{ marker }}</Marker>
{% endif %}
<VirtualMFADevices>
{% for device in devices %}
<member>
<SerialNumber>{{ device.serial_number }}</SerialNumber>
{% if device.enable_date %}
<EnableDate>{{ device.enabled_iso_8601 }}</EnableDate>
{% endif %}
{% if device.user %}
<User>
<Path>{{ device.user.path }}</Path>
<UserName>{{ device.user.name }}</UserName>
<UserId>{{ device.user.id }}</UserId>
<CreateDate>{{ device.user.created_iso_8601 }}</CreateDate>
<Arn>{{ device.user.arn }}</Arn>
</User>
{% endif %}
</member>
{% endfor %}
</VirtualMFADevices>
</ListVirtualMFADevicesResult>
<ResponseMetadata>
<RequestId>b61ce1b1-0401-11e1-b2f8-2dEXAMPLEbfc</RequestId>
</ResponseMetadata>
</ListVirtualMFADevicesResponse>"""
LIST_ACCOUNT_ALIASES_TEMPLATE = """<ListAccountAliasesResponse xmlns="https://iam.amazonaws.com/doc/2010-05-08/">
<ListAccountAliasesResult>
<IsTruncated>false</IsTruncated>

View File

@ -753,6 +753,263 @@ def test_mfa_devices():
len(response['MFADevices']).should.equal(0)
@mock_iam
def test_create_virtual_mfa_device():
client = boto3.client('iam', region_name='us-east-1')
response = client.create_virtual_mfa_device(
VirtualMFADeviceName='test-device'
)
device = response['VirtualMFADevice']
device['SerialNumber'].should.equal('arn:aws:iam::123456789012:mfa/test-device')
device['Base32StringSeed'].decode('ascii').should.match('[A-Z234567]')
device['QRCodePNG'].should_not.be.empty
response = client.create_virtual_mfa_device(
Path='/',
VirtualMFADeviceName='test-device-2'
)
device = response['VirtualMFADevice']
device['SerialNumber'].should.equal('arn:aws:iam::123456789012:mfa/test-device-2')
device['Base32StringSeed'].decode('ascii').should.match('[A-Z234567]')
device['QRCodePNG'].should_not.be.empty
response = client.create_virtual_mfa_device(
Path='/test/',
VirtualMFADeviceName='test-device'
)
device = response['VirtualMFADevice']
device['SerialNumber'].should.equal('arn:aws:iam::123456789012:mfa/test/test-device')
device['Base32StringSeed'].decode('ascii').should.match('[A-Z234567]')
device['QRCodePNG'].should_not.be.empty
@mock_iam
def test_create_virtual_mfa_device_errors():
client = boto3.client('iam', region_name='us-east-1')
client.create_virtual_mfa_device(
VirtualMFADeviceName='test-device'
)
client.create_virtual_mfa_device.when.called_with(
VirtualMFADeviceName='test-device'
).should.throw(
ClientError,
'MFADevice entity at the same path and name already exists.'
)
client.create_virtual_mfa_device.when.called_with(
Path='test',
VirtualMFADeviceName='test-device'
).should.throw(
ClientError,
'The specified value for path is invalid. '
'It must begin and end with / and contain only alphanumeric characters and/or / characters.'
)
client.create_virtual_mfa_device.when.called_with(
Path='/test//test/',
VirtualMFADeviceName='test-device'
).should.throw(
ClientError,
'The specified value for path is invalid. '
'It must begin and end with / and contain only alphanumeric characters and/or / characters.'
)
too_long_path = '/{}/'.format('b' * 511)
client.create_virtual_mfa_device.when.called_with(
Path=too_long_path,
VirtualMFADeviceName='test-device'
).should.throw(
ClientError,
'1 validation error detected: '
'Value "{}" at "path" failed to satisfy constraint: '
'Member must have length less than or equal to 512'
)
@mock_iam
def test_delete_virtual_mfa_device():
client = boto3.client('iam', region_name='us-east-1')
response = client.create_virtual_mfa_device(
VirtualMFADeviceName='test-device'
)
serial_number = response['VirtualMFADevice']['SerialNumber']
client.delete_virtual_mfa_device(
SerialNumber=serial_number
)
response = client.list_virtual_mfa_devices()
response['VirtualMFADevices'].should.have.length_of(0)
response['IsTruncated'].should_not.be.ok
@mock_iam
def test_delete_virtual_mfa_device_errors():
client = boto3.client('iam', region_name='us-east-1')
serial_number = 'arn:aws:iam::123456789012:mfa/not-existing'
client.delete_virtual_mfa_device.when.called_with(
SerialNumber=serial_number
).should.throw(
ClientError,
'VirtualMFADevice with serial number {0} doesn\'t exist.'.format(serial_number)
)
@mock_iam
def test_list_virtual_mfa_devices():
client = boto3.client('iam', region_name='us-east-1')
response = client.create_virtual_mfa_device(
VirtualMFADeviceName='test-device'
)
serial_number_1 = response['VirtualMFADevice']['SerialNumber']
response = client.create_virtual_mfa_device(
Path='/test/',
VirtualMFADeviceName='test-device'
)
serial_number_2 = response['VirtualMFADevice']['SerialNumber']
response = client.list_virtual_mfa_devices()
response['VirtualMFADevices'].should.equal([
{
'SerialNumber': serial_number_1
},
{
'SerialNumber': serial_number_2
}
])
response['IsTruncated'].should_not.be.ok
response = client.list_virtual_mfa_devices(
AssignmentStatus='Assigned'
)
response['VirtualMFADevices'].should.have.length_of(0)
response['IsTruncated'].should_not.be.ok
response = client.list_virtual_mfa_devices(
AssignmentStatus='Unassigned'
)
response['VirtualMFADevices'].should.equal([
{
'SerialNumber': serial_number_1
},
{
'SerialNumber': serial_number_2
}
])
response['IsTruncated'].should_not.be.ok
response = client.list_virtual_mfa_devices(
AssignmentStatus='Any',
MaxItems=1
)
response['VirtualMFADevices'].should.equal([
{
'SerialNumber': serial_number_1
}
])
response['IsTruncated'].should.be.ok
response['Marker'].should.equal('1')
response = client.list_virtual_mfa_devices(
AssignmentStatus='Any',
Marker=response['Marker']
)
response['VirtualMFADevices'].should.equal([
{
'SerialNumber': serial_number_2
}
])
response['IsTruncated'].should_not.be.ok
@mock_iam
def test_list_virtual_mfa_devices_errors():
client = boto3.client('iam', region_name='us-east-1')
client.create_virtual_mfa_device(
VirtualMFADeviceName='test-device'
)
client.list_virtual_mfa_devices.when.called_with(
Marker='100'
).should.throw(
ClientError,
'Invalid Marker.'
)
@mock_iam
def test_enable_virtual_mfa_device():
client = boto3.client('iam', region_name='us-east-1')
response = client.create_virtual_mfa_device(
VirtualMFADeviceName='test-device'
)
serial_number = response['VirtualMFADevice']['SerialNumber']
client.create_user(UserName='test-user')
client.enable_mfa_device(
UserName='test-user',
SerialNumber=serial_number,
AuthenticationCode1='234567',
AuthenticationCode2='987654'
)
response = client.list_virtual_mfa_devices(
AssignmentStatus='Unassigned'
)
response['VirtualMFADevices'].should.have.length_of(0)
response['IsTruncated'].should_not.be.ok
response = client.list_virtual_mfa_devices(
AssignmentStatus='Assigned'
)
device = response['VirtualMFADevices'][0]
device['SerialNumber'].should.equal(serial_number)
device['User']['Path'].should.equal('/')
device['User']['UserName'].should.equal('test-user')
device['User']['UserId'].should_not.be.empty
device['User']['Arn'].should.equal('arn:aws:iam::123456789012:user/test-user')
device['User']['CreateDate'].should.be.a(datetime)
device['EnableDate'].should.be.a(datetime)
response['IsTruncated'].should_not.be.ok
client.deactivate_mfa_device(
UserName='test-user',
SerialNumber=serial_number
)
response = client.list_virtual_mfa_devices(
AssignmentStatus='Assigned'
)
response['VirtualMFADevices'].should.have.length_of(0)
response['IsTruncated'].should_not.be.ok
response = client.list_virtual_mfa_devices(
AssignmentStatus = 'Unassigned'
)
response['VirtualMFADevices'].should.equal([
{
'SerialNumber': serial_number
}
])
response['IsTruncated'].should_not.be.ok
@mock_iam_deprecated()
def test_delete_user_deprecated():
conn = boto.connect_iam()