Various fixes and update to Python 3.10

Author: jplock

dependencies/requirements.txt

@@ -1 +1 @@
-aws-lambda-powertools[tracer,validation,aws-sdk]==2.11.0
+aws-lambda-powertools[tracer,validation,aws-sdk]==2.14.1

mypy.ini

@@ -0,0 +1,28 @@
+[mypy]
+warn_return_any=False
+warn_unused_configs=True
+no_implicit_optional=True
+warn_redundant_casts=True
+warn_unused_ignores=True
+show_column_numbers = True
+show_error_codes = True
+show_error_context = True
+disable_error_code = annotation-unchecked
+
+[mypy-boto3]
+ignore_missing_imports = True
+
+[mypy-botocore]
+ignore_missing_imports = True
+
+[mypy-botocore.response]
+ignore_missing_imports = True
+
+[mypy-botocore.config]
+ignore_missing_imports = True
+
+[mypy-botocore.compat]
+ignore_missing_imports = True
+
+[mypy-botocore.exceptions]
+ignore_missing_imports = True

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.black]
-line-length = 88
-target-version = ['py39']
+line-length = 120
+target-version = ['py310']
 include = '\.pyi?$'
 extend-exclude = '''
 (

requirements-dev.txt

@@ -1,3 +1,11 @@
 black==23.3.0
 wheel==0.40.0
-pre-commit==3.2.1
+pre-commit==3.2.2
+mypy-boto3-ec2==1.26.106
+mypy-boto3-ecs==1.26.116
+mypy-boto3-iam==1.26.97
+mypy-boto3-identitystore==1.26.107
+mypy-boto3-organizations==1.26.83
+mypy-boto3-sts==1.26.57
+mypy-boto3-servicecatalog==1.26.109
+mypy-boto3-sso-admin==1.26.63

src/regional/account_setup/lambda_handler.py

@@ -37,21 +37,24 @@
 @tracer.capture_lambda_handler
 @logger.inject_lambda_context(log_event=True)
 def handler(event: Dict[str, Any], context: LambdaContext) -> None:
-
     account_id = event["account_id"]
     region_name = event["region"]
 
     logger.append_keys(account_id=account_id, region=region_name)
+    tracer.put_annotation("AccountId", account_id)
+    tracer.put_annotation("Region", region_name)
 
     session = boto3.Session()
 
     assumed_session = STS(session).assume_role(account_id)
 
-    logger.info(f"Deleting default VPC from {region_name} in {account_id}")
     ec2 = EC2(assumed_session, region_name)
     default_vpc_id = ec2.get_default_vpc_id()
     if default_vpc_id:
+        logger.info(f"Deleting default VPC {default_vpc_id} from {region_name} in {account_id}")
         ec2.delete_vpc(default_vpc_id)
+    else:
+        logger.debug(f"No default VPC found in {region_name} in {account_id}")
 
     logger.info(f"Setting default ECS settings in {region_name} in {account_id}")
     ecs = ECS(assumed_session, region_name)

src/regional/account_setup/resources/ec2.py

@@ -19,19 +19,22 @@
 * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 """
 
-from typing import Optional
+from typing import Optional, TYPE_CHECKING
 
 from aws_lambda_powertools import Logger
 import boto3
 
+if TYPE_CHECKING:
+    from mypy_boto3_ec2 import EC2Client, EC2ServiceResource
+
 logger = Logger(child=True)
 
 __all__ = ["EC2"]
 
 
 class EC2:
     def __init__(self, session: boto3.Session, region: str) -> None:
-        self.client = session.client("ec2", region_name=region)
+        self.client: EC2Client = session.client("ec2", region_name=region)
         self.session = session
         self.region_name = region
 
@@ -43,11 +46,11 @@ def get_default_vpc_id(self) -> Optional[str]:
             if vpc["IsDefault"]:
                 return vpc["VpcId"]
 
-        logger.debug("No default VPC found")
+        logger.debug(f"No default VPC found in {self.region_name}", region=self.region_name)
         return None
 
     def delete_vpc(self, vpc_id: str) -> None:
-        ec2 = self.session.resource("ec2", region_name=self.region_name)
+        ec2: EC2ServiceResource = self.session.resource("ec2", region_name=self.region_name)
         vpc = ec2.Vpc(vpc_id)
 
         # detach and delete all gateways associated with the vpc
@@ -78,12 +81,16 @@ def delete_vpc(self, vpc_id: str) -> None:
                 nacl.delete()
 
         # DHCP Options
-        if vpc.dhcp_options:
-            vpc.associate_dhcp_options(
-                DhcpOptionsId="default"
-            )  # associate no DHCP options
-            vpc.dhcp_options.delete()
+        if vpc.dhcp_options and vpc.dhcp_options_id != "default":
+            dhcp_options_id = vpc.dhcp_options_id
+
+            vpc.associate_dhcp_options(DhcpOptionsId="default")  # associate no DHCP options
+
+            dhcp_options = ec2.DhcpOptions(dhcp_options_id)
+            dhcp_options.delete()
 
         # Delete VPC
         self.client.delete_vpc(VpcId=vpc_id)
-        logger.info(f"VPC {vpc_id} and associated resources has been deleted.")
+        logger.info(
+            f"VPC {vpc_id} and associated resources has been deleted in {self.region_name}.", region=self.region_name
+        )

src/regional/account_setup/resources/ecs.py

@@ -19,14 +19,24 @@
 * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 """
 
+from typing import TYPE_CHECKING
+
+from aws_lambda_powertools import Logger
 import boto3
+import botocore
+
+if TYPE_CHECKING:
+    from mypy_boto3_ecs import ECSClient
+
+logger = Logger(child=True)
 
 __all__ = ["ECS"]
 
 
 class ECS:
     def __init__(self, session: boto3.Session, region: str) -> None:
-        self.client = session.client("ecs", region_name=region)
+        self.client: ECSClient = session.client("ecs", region_name=region)
+        self.region = region
 
     def put_account_setting_default(self) -> None:
         names = [
@@ -35,7 +45,16 @@ def put_account_setting_default(self) -> None:
             "containerInstanceLongArnFormat",
             "awsvpcTrunking",
             "containerInsights",
-            # "dualStackIPv6", # not yet supported
+            "dualStackIPv6",
         ]
         for name in names:
-            self.client.put_account_setting_default(name=name, value="enabled")
+            try:
+                self.client.put_account_setting_default(name=name, value="enabled")
+            except botocore.exceptions.ClientError:
+                logger.exception(f"Unable to enable ECS setting {name} in {self.region}")
+
+        # documentation on https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ecs-account-settings.html#tag-resources-setting is incorrect
+        try:
+            self.client.put_account_setting_default(name="tagResourceAuthorization", value="on")
+        except botocore.exceptions.ClientError:
+            logger.exception(f"Unable to enable ECS setting tagResourceAuthorization in {self.region}")

src/regional/account_setup/resources/sts.py

@@ -20,10 +20,14 @@
 """
 
 import os
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools import Logger
 import boto3
 
+if TYPE_CHECKING:
+    from mypy_boto3_sts import STSClient
+
 logger = Logger(child=True)
 EXECUTION_ROLE_NAME = os.environ["EXECUTION_ROLE_NAME"]
 AWS_PARTITION = os.environ["AWS_PARTITION"]
@@ -33,11 +37,9 @@
 
 class STS:
     def __init__(self, session: boto3.Session) -> None:
-        self.client = session.client("sts")
+        self.client: STSClient = session.client("sts")
 
-    def assume_role(
-        self, account_id: str, role_session_name: str = "AccountSetup"
-    ) -> boto3.Session:
+    def assume_role(self, account_id: str, role_session_name: str = "AccountSetup") -> boto3.Session:
         """
         Assume the AWSControlTowerExecution role in an account
         """

src/service_catalog_portfolio/account_setup/lambda_handler.py

@@ -20,7 +20,7 @@
 """
 
 import os
-from typing import Dict, Any, List
+from typing import Dict, Any, List, Optional
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.utilities.typing import LambdaContext
@@ -46,8 +46,7 @@ def get_env_list(key: str) -> List[str]:
 @tracer.capture_lambda_handler
 @logger.inject_lambda_context(log_event=True)
 def handler(event: Dict[str, Any], context: LambdaContext) -> None:
-
-    account_id = event.get("account", {}).get("accountId")
+    account_id: Optional[str] = event.get("account", {}).get("accountId")
     if not account_id:
         raise Exception("Account ID not found in event")
 
@@ -73,11 +72,7 @@ def handler(event: Dict[str, Any], context: LambdaContext) -> None:
         to_remove = existing_principals - role_arns
 
         for role_arn in to_add:
-            servicecatalog.associate_principal_with_portfolio(
-                portfolio_id=portfolio_id, principal_arn=role_arn
-            )
+            servicecatalog.associate_principal_with_portfolio(portfolio_id=portfolio_id, principal_arn=role_arn)
 
         for role_arn in to_remove:
-            servicecatalog.disassociate_principal_from_portfolio(
-                portfolio_id=portfolio_id, principal_arn=role_arn
-            )
+            servicecatalog.disassociate_principal_from_portfolio(portfolio_id=portfolio_id, principal_arn=role_arn)

src/service_catalog_portfolio/account_setup/resources/iam.py

@@ -20,19 +20,24 @@
 """
 
 from functools import lru_cache
-from typing import Optional, Dict
+from typing import Optional, Dict, TYPE_CHECKING
 
 import boto3
 
+if TYPE_CHECKING:
+    from mypy_boto3_iam import IAMClient
+
 __all__ = ["IAM"]
 
 AWS_SSO_ROLE_PREFIX = "AWSReservedSSO_"
 
 
 class IAM:
-    def __init__(self, session: boto3.Session) -> None:
-        self.client = session.client("iam")
-        self._roles = {}
+    def __init__(self, session: Optional[boto3.Session] = None) -> None:
+        if not session:
+            session = boto3._get_default_session()
+        self.client: IAMClient = session.client("iam")
+        self._roles: Dict[str, str] = {}
 
     def get_sso_roles(self) -> Dict[str, str]:
         """
@@ -48,11 +53,7 @@ def get_sso_roles(self) -> Dict[str, str]:
             for role in page.get("Roles", []):
                 if role["RoleName"].startswith(AWS_SSO_ROLE_PREFIX):
                     # AWSReservedSSO_AWSAdministratorAccess_a1ff75f56dfb0e2f -> AWSAdministratorAccess
-                    permission_set_name = (
-                        role["RoleName"]
-                        .rsplit("_", 1)[0]
-                        .replace(AWS_SSO_ROLE_PREFIX, "")
-                    )
+                    permission_set_name = role["RoleName"].rsplit("_", 1)[0].replace(AWS_SSO_ROLE_PREFIX, "")
                     roles[permission_set_name] = role["Arn"]
 
         self._roles = roles

src/service_catalog_portfolio/account_setup/resources/servicecatalog.py

@@ -19,35 +19,34 @@
 * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 """
 
-from typing import Set
+from typing import Set, TYPE_CHECKING, Optional
 
 from aws_lambda_powertools import Logger
 import boto3
 import botocore
 
+if TYPE_CHECKING:
+    from mypy_boto3_servicecatalog import ServiceCatalogClient, ListPrincipalsForPortfolioPaginator
+
 logger = Logger(child=True)
 
 __all__ = ["ServiceCatalog"]
 
 
 class ServiceCatalog:
-    def __init__(self, session: boto3.Session) -> None:
+    def __init__(self, session: Optional[boto3.Session] = None) -> None:
         if not session:
             session = boto3._get_default_session()
-        self.client = session.client("servicecatalog")
+        self.client: ServiceCatalogClient = session.client("servicecatalog")
 
     def accept_portfolio_share(self, portfolio_id: str) -> None:
         try:
-            self.client.accept_portfolio_share(
-                PortfolioId=portfolio_id, PortfolioShareType="AWS_ORGANIZATIONS"
-            )
+            self.client.accept_portfolio_share(PortfolioId=portfolio_id, PortfolioShareType="AWS_ORGANIZATIONS")
         except botocore.exceptions.ClientError:
             logger.exception("Unable to accept portfolio share")
             raise
 
-    def associate_principal_with_portfolio(
-        self, portfolio_id: str, principal_arn: str
-    ) -> None:
+    def associate_principal_with_portfolio(self, portfolio_id: str, principal_arn: str) -> None:
         try:
             self.client.associate_principal_with_portfolio(
                 PortfolioId=portfolio_id,
@@ -58,13 +57,9 @@ def associate_principal_with_portfolio(
             logger.exception("Unable to associate princpal with portfolio")
             raise
 
-    def disassociate_principal_from_portfolio(
-        self, portfolio_id: str, principal_arn: str
-    ) -> None:
+    def disassociate_principal_from_portfolio(self, portfolio_id: str, principal_arn: str) -> None:
         try:
-            self.client.disassociate_principal_from_portfolio(
-                PortfolioId=portfolio_id, PrincipalARN=principal_arn
-            )
+            self.client.disassociate_principal_from_portfolio(PortfolioId=portfolio_id, PrincipalARN=principal_arn)
         except botocore.exceptions.ClientError as error:
             if error.response["Error"]["Code"] != "ResourceNotFoundException":
                 logger.exception("Unable to disassociate princpal from portfolio")
@@ -73,7 +68,7 @@ def disassociate_principal_from_portfolio(
     def list_principals_for_portfolio(self, portfolio_id: str) -> Set[str]:
         principals = set()
 
-        paginator = self.client.get_paginator("list_principals_for_portfolio")
+        paginator: ListPrincipalsForPortfolioPaginator = self.client.get_paginator("list_principals_for_portfolio")
         page_iterator = paginator.paginate(PortfolioId=portfolio_id)
         for page in page_iterator:
             for principal in page.get("Principals", []):

src/service_catalog_portfolio/account_setup/resources/sts.py

@@ -20,20 +20,24 @@
 """
 
 import os
+from typing import TYPE_CHECKING, Optional
 
 import boto3
 
+if TYPE_CHECKING:
+    from mypy_boto3_sts import STSClient
+
 EXECUTION_ROLE_NAME = os.environ["EXECUTION_ROLE_NAME"]
 AWS_PARTITION = os.environ["AWS_PARTITION"]
 
 __all__ = ["STS"]
 
 
 class STS:
-    def __init__(self, session: boto3.Session = None) -> None:
+    def __init__(self, session: Optional[boto3.Session] = None) -> None:
         if not session:
             session = boto3._get_default_session()
-        self.client = session.client("sts")
+        self.client: STSClient = session.client("sts")
 
     def assume_role(self, account_id: str, role_session_name: str) -> boto3.Session:
         """

src/sso_assignment/account_setup/constants.py

@@ -19,8 +19,6 @@
 * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 """
 
-__all__ = ["GROUP_ACCOUNT_PREFIX", "GROUP_ORG_PREFIX"]
-
 GROUP_ACCOUNT_PREFIX = "AWS-A-"
 
 GROUP_ORG_PREFIX = "AWS-O-"