#!/usr/bin/env python

"""

Script to gather all CloudFormation resource types available across AWS regions.

The collected data is written to `localstack/services/cloudformation/resources.py` as

* `AWS_AVAILABLE_CFN_RESOURCES`: a dict, mapping resource names to the regions they are available in.

* `AWS_CFN_REGIONS_SCANNED`: a set of regions for which listing CloudFormation types succeeded.

The script expects valid AWS credentials either via the environment or configured profiles (for local execution).

Note:

We evaluated scraping the public documentation/zip schema endpoints, but they do not have resources like

``AWS::OpsWorksCM::Server`` that are still returned by the CloudFormation API.

To make sure we cover all resources, we rely on the list of available types from the API.

"""

import argparse

import sys

from collections.abc import Iterable

from pathlib import Path

import boto3

from botocore.exceptions import ClientError

REPO_ROOT = Path(__file__).resolve().parents[1]

DEFAULT_RESOURCE_FILE = (

REPO_ROOT / "localstack-core" / "localstack" / "services" / "cloudformation" / "resources.py"

)

def get_regions(

session: boto3.session.Session, explicit_regions: Iterable[str] | None = None

) -> list[str]:

if explicit_regions:

return sorted(set(explicit_regions))

return sorted(session.get_available_regions("cloudformation"))

def collect_region_resource_types(

session: boto3.session.Session, region: str

) -> tuple[set[str], bool]:

client = session.client("cloudformation", region_name=region)

resources: set[str] = set()

token: str | None = None

succeeded = True

while True:

try:

params = {

"Visibility": "PUBLIC",

"Type": "RESOURCE",

"Filters": {"Category": "AWS_TYPES"},

}

if token:

params["NextToken"] = token

response = client.list_types(**params)

except ClientError as exc:

print(f"Skipping region {region} due to error while listing types: {exc}")

succeeded = False

break

for summary in response.get("TypeSummaries", []):

type_name = summary.get("TypeName")

if type_name:

resources.add(type_name)

token = response.get("NextToken")

if not token:

break

return resources, succeeded

def collect_all_resource_types(

session: boto3.session.Session, regions: Iterable[str]

) -> tuple[dict[str, set[str]], set[str]]:

aggregated: dict[str, set[str]] = {}

successful_regions: set[str] = set()

for region in regions:

print(f"Collecting CloudFormation resource types in region {region}")

region_resources, succeeded = collect_region_resource_types(session, region)

if not succeeded:

continue

successful_regions.add(region)

for resource in region_resources:

aggregated.setdefault(resource, set()).add(region)

return aggregated, successful_regions

def render_resource_file(resources: dict[str, set[str]], successful_regions: set[str]) -> str:

lines: list[str] = [

'"""Generated by scripts/update_cfn_resources.py – do not edit manually."""',

"",

]

if resources:

lines.append("AWS_AVAILABLE_CFN_RESOURCES = {")

for resource in sorted(resources):

regions = sorted(resources[resource])

if regions:

lines.append(f' "{resource}": [')

for region_name in regions:

lines.append(f' "{region_name}",')

lines.append(" ],")

else:

lines.append(f' "{resource}": [],')

lines.append("}")

else:

lines.append("AWS_AVAILABLE_CFN_RESOURCES = {}")

lines.append("")

if successful_regions:

lines.append("AWS_CFN_REGIONS_SCANNED = {")

for region in sorted(successful_regions):

lines.append(f' "{region}",')

lines.append("}")

else:

lines.append("AWS_CFN_REGIONS_SCANNED = set()")

lines.append("")

return "\n".join(lines)

def write_resource_file(path: Path, content: str) -> None:

path.write_text(content, encoding="utf-8")

def parse_args() -> argparse.Namespace:

parser = argparse.ArgumentParser(description="Update the AWS_AVAILABLE_CFN_RESOURCES constant.")

parser.add_argument("--profile", help="AWS profile name to use")

parser.add_argument(

"--regions",

nargs="*",

help="Optional list of AWS regions to scan. Defaults to all regions supporting CloudFormation.",

)

parser.add_argument(

"--resource-file",

default=str(DEFAULT_RESOURCE_FILE),

help="Path to the resources.py file that should be updated.",

)

parser.add_argument(

"--dry-run", action="store_true", help="Do not write the file, only print the resources."

)

return parser.parse_args()

def main() -> int:

args = parse_args()

session_kwargs = {}

if args.profile:

session_kwargs["profile_name"] = args.profile

try:

session = boto3.session.Session(**session_kwargs)

except Exception as exc:

print(f"Failed to create boto3 session: {exc}")

return 1

regions = get_regions(session, args.regions)

if not regions:

print("Could not determine any regions to scan.")

return 1

print(f"Scanning CloudFormation resource types in {len(regions)} regions")

resources, successful_regions = collect_all_resource_types(session, regions)

if not resources:

print("No CloudFormation resources were discovered.")

return 1

print(f"Collected {len(resources)} resources across {len(successful_regions)} regions:")

print("Updating resource file...")

content = render_resource_file(resources, successful_regions)

if args.dry_run:

sys.stdout.write(content)

return 0

resource_file = Path(args.resource_file)

write_resource_file(resource_file, content)

print(

f"Updated {resource_file} with {len(resources)} CloudFormation resource types "

f"across {len(successful_regions)} regions."

)

return 0

if __name__ == "__main__":

sys.exit(main())