Add ability to delete key pairs

ocw/lib/ec2.py

@@ -369,3 +369,15 @@ def cleanup_images(self, valid_period_days: float) -> None:
                         else:
                             self.log_info(f"Delete image '{img['Name']}' (ami:{img['ImageId']})")
                             self.ec2_client(region).deregister_image(ImageId=img['ImageId'], DryRun=False)
+
+    def cleanup_keypairs(self):
+        self.log_dbg('Call cleanup_images')
+        for region in self.all_regions:
+            response = self.ec2_client(region).describe_key_pairs()
+            self.log_dbg(f"Found {len(response['KeyPairs'])} images in {region}")
+            for keypair in response['KeyPairs']:
+                if self.dry_run:
+                    self.log_info(f"KeyPair deletion {keypair['KeyName']} skipped due to dry run mode")
+                else:
+                    self.log_info(f"Delete KeyPair '{keypair['KeyName']}'")
+                    self.ec2_client(region).delete_key_pair(KeyName=keypair['KeyName'])

ocw/management/commands/cleanup_ec2_keypairs.py

@@ -0,0 +1,11 @@
+from django.core.management.base import BaseCommand
+from webui.PCWConfig import PCWConfig
+from ocw.lib.ec2 import EC2
+
+
+class Command(BaseCommand):
+    help = 'Delete all leftovers in all providers (according to pcw.ini)'
+
+    def handle(self, *args, **options):
+        for namespace in PCWConfig.get_namespaces_for('cleanup'):
+            EC2(namespace).cleanup_keypairs()

tests/test_ec2.py

@@ -89,6 +89,7 @@ class MockedEC2Client():
     response = {}
     deleted_images = list()
     deleted_volumes = list()
+    deleted_keys = list()
     snapshotid_to_delete = 'delete_me'
     volumeid_to_delete = 'delete_me'
     snapshotid_i_have_ami = 'you_can_not_delete_me'
@@ -139,6 +140,9 @@ def describe_snapshots(self, *args, **kwargs):
     def deregister_image(self, *args, **kwargs):
         MockedEC2Client.deleted_images.append(kwargs['ImageId'])
 
+    def delete_key_pair(self, *args, **kwargs):
+        MockedEC2Client.deleted_keys.append(kwargs['KeyName'])
+
     def delete_snapshot(self, SnapshotId):
         if MockedEC2Client.delete_snapshot_raise_error:
             error_response = {'Error': {'Code': 'InvalidSnapshot.InUse', 'Message': 'Message'}}
@@ -155,6 +159,9 @@ def describe_volumes(self, *args, **kwargs):
     def describe_vpcs(self, Filters):
         return MockedEC2Client.response
 
+    def describe_key_pairs(self):
+        return MockedEC2Client.response
+
     def describe_vpc_endpoints(self, Filters):
         return MockedEC2Client.response
 
@@ -301,6 +308,16 @@ def test_cleanup_snapshots_cleanup_all_new(ec2_patch):
     assert len(MockedEC2Client.ec2_snapshots) == 2
 
 
+def test_cleanup_keypairs(ec2_patch):
+    MockedEC2Client.response = {'KeyPairs': [{'KeyName': 'key1'}]}
+    ec2_patch.dry_run = True
+    ec2_patch.cleanup_keypairs()
+    assert len(MockedEC2Client.deleted_keys) == 0
+    ec2_patch.dry_run = False
+    ec2_patch.cleanup_keypairs()
+    assert len(MockedEC2Client.deleted_keys) == 1
+
+
 def test_cleanup_snapshots_cleanup_one_old(ec2_patch):
     MockedEC2Client.response = {
         'Snapshots': [{'SnapshotId': MockedEC2Client.snapshotid_to_delete, 'StartTime': older_than_max_age_date}]