mypy

Author: Justin

aws_sra_examples/solutions/shield_advanced/shield_advanced/lambda/src/common.py

@@ -183,67 +183,3 @@ def get_enabled_regions(customer_regions: str, control_tower_regions_only: bool
         }
     )
     return enabled_regions
-
-def parse_account_id_from_iam_role_arn(role_arn) -> str:
-    """Takes a role arn and returns the account number
-
-    Args:
-        role_arn: role's arn
-
-    Returns:
-        12 digit aws account number
-    """
-    pattern = re.compile('arn:aws:iam::(\d+):')
-    match = pattern.search(role_arn)
-    account_number = match.group(1)
-    return account_number
-
-def create_service_linked_role(
-    service_linked_role_name: str,
-    service_name: str,
-    description: str = "",
-    iam_client: IAMClient = None,
-):
-    """Create the service linked role, if it does not exist.
-
-    Args:
-        service_linked_role_name: Service Linked Role Name
-        service_name: AWS Service Name
-        description: Description
-        iam_client: IAMClient
-    """
-    if not iam_client:
-        iam_client = boto3.client("iam")
-    try:
-        response = iam_client.get_role(RoleName=service_linked_role_name)
-        api_call_details = {"API_Call": "iam:GetRole", "API_Response": response}
-        LOGGER.info(api_call_details)
-    except iam_client.exceptions.NoSuchEntityException:
-        iam_client.create_service_linked_role(AWSServiceName=service_name, Description=description)
-
-
-def snake_to_camel(snake_str: str) -> str:
-    """Convert snake case to camel case.
-
-    Args:
-        snake_str: String to convert
-
-    Returns:
-        Camel case string
-    """
-    camel_str = snake_str.title().replace("_", "")
-    return camel_str[0].lower() + camel_str[1:]
-
-
-def camel_to_snake_upper(camel_str: str) -> str:
-    """Concert camel case to snake upper case.
-
-    Args:
-        camel_str: String to convert
-
-    Returns:
-        Snake upper case string
-    """
-    snake_chars = ["_" + x.lower() if x.isupper() else x for x in camel_str]
-    snake_str = "".join(snake_chars).lstrip("_")
-    return snake_str.upper()

aws_sra_examples/solutions/shield_advanced/shield_advanced/lambda/src/shield.py

@@ -26,12 +26,10 @@
     from mypy_boto3_s3 import S3Client
     from mypy_boto3_shield import ShieldClient
     from mypy_boto3_shield.type_defs import (
-        AssociateProactiveEngagementDetailsRequestRequestTypeDef,
         CreateProtectionResponseTypeDef,
         DescribeEmergencyContactSettingsResponseTypeDef,
         DescribeProtectionResponseTypeDef,
         DescribeSubscriptionResponseTypeDef,
-        DisableApplicationLayerAutomaticResponseRequestRequestTypeDef,
         EmergencyContactTypeDef,
         ProtectionTypeDef,
     )
@@ -340,7 +338,7 @@ def delete_drt_role(account_session: boto3.Session, role_name: str) -> None:
         LOGGER.info(api_call_details)
     except iam_client.exceptions.NoSuchEntityException as nse:
         LOGGER.info(f"NoSuchEntityException {nse}")
-        LOGGER.info(f"Continuing...")
+        LOGGER.info("Continuing...")
 
 
 def check_if_role_exists(iam_client: IAMClient, role_name: str) -> str:
@@ -523,8 +521,10 @@ def check_proactive_engagement_enabled(shield_client: ShieldClient, params: dict
         elif proactive_engagement_status == "PENDING":
             time.sleep(5)
             check_proactive_engagement_enabled(shield_client, params, retry + 1)
+        return False
     else:
         return False
+    return False
 
 
 def check_if_protection_group_exists(shield_client: ShieldClient, protection_group_id: str) -> bool:
@@ -590,7 +590,7 @@ def update_protection_group(
         "APPLICATION_LOAD_BALANCER",
         "GLOBAL_ACCELERATOR",
     ],
-    pg_members: list[str],
+    pg_members: str,
 ) -> None:
     """Updates an existing protection group
 
@@ -637,7 +637,7 @@ def create_protection_group(shield_client: ShieldClient, params: dict, account_i
             "APPLICATION_LOAD_BALANCER",
             "GLOBAL_ACCELERATOR",
         ] = params[f"PROTECTION_GROUP_{i}_RESOURCE_TYPE"]
-        pg_members: list = params[f"PROTECTION_GROUP_{i}_MEMBERS"]
+        pg_members: str = params[f"PROTECTION_GROUP_{i}_MEMBERS"]
         if pg_id != "" and pg_account_id == account_id:
             if check_if_protection_group_exists(shield_client, pg_id):
                 LOGGER.info(f"Protection_Group_{i} already exists in {account_id}")
@@ -728,9 +728,7 @@ def disable_proactive_engagement(shield_client: ShieldClient) -> None:
         shield_client: shield client
     """
     try:
-        disable_proactive_engagement_response: DisableApplicationLayerAutomaticResponseRequestRequestTypeDef = (
-            shield_client.disable_proactive_engagement()
-        )
+        disable_proactive_engagement_response = shield_client.disable_proactive_engagement()
         api_call_details = {"API_Call": "shield:DisableProactiveEngagement", "API_Response": disable_proactive_engagement_response}
         LOGGER.info(api_call_details)
     except ClientError as e: