PK!@YSStroposphere_ez/__init__.pyfrom troposphere_ez._stack_manager_base import StackManagerBase, IStackManagerBase PK!%troposphere_ez/_stack_manager_base.pyimport json import os import os.path import shlex import subprocess from abc import ABC, abstractmethod from collections import OrderedDict from tempfile import TemporaryDirectory from termcolor import cprint from typing import List, Optional from troposphere import Output, Template class IStackManagerBase(ABC): """StackManagerBaseのインターフェース """ aws_profile: str stage: str prefix: str @abstractmethod def rec_to_output(self, title, value): pass @abstractmethod def add_resource(self, res): pass @abstractmethod def toyaml(self): pass @abstractmethod def apply(self): pass @property @abstractmethod def output_results(self) -> OrderedDict: pass @abstractmethod def pprint_output_results(self): pass @abstractmethod def run_aws_cli(self, *rest_args, quiet=True): pass @abstractmethod def run_aws_cli_skip_err(self, *rest_args, quiet=False) -> Optional[str]: pass class StackManagerBase(IStackManagerBase): aws_profile: str stage: str prefix: str _template: Template _outputs: List[Output] _output_results: Optional[OrderedDict] def __init__(self, aws_profile: str, stage: str, prefix: str): self.aws_profile = aws_profile self.stage = stage self.prefix = prefix self._template = Template() self._outputs = [] self._output_results = None def rec_to_output(self, title, value): self._outputs.append(Output(title, Value=value)) def add_resource(self, res): self._template.add_resource(res) def toyaml(self): self._template.add_output(self._outputs) return self._template.to_yaml().strip() def apply(self): self._template.add_output(self._outputs) with TemporaryDirectory() as tmpdir: the_file = os.path.join(tmpdir, "template.yml") with open(the_file, "wt") as fp: fp.write(self._template.to_yaml()) cmd_env = os.environ.copy() # AWS_SDK_LOAD_CONFIG をセットしていないと~/.aws/configのregsionを読んでくれない cmd_env["AWS_SDK_LOAD_CONFIG"] = "true" cmd_args = [ "cfn-create-or-update", "--profile", self.aws_profile, "--template-body", f"file://{the_file}", "--capabilities", "CAPABILITY_NAMED_IAM", "--stack-name", f"{self.prefix}-{self.stage}", "--wait", ] print("$ ", end="") cprint(" ".join([shlex.quote(s) for s in cmd_args]), "cyan") subprocess.check_output(cmd_args, env=cmd_env) self._sync_output_results() def run_aws_cli(self, *rest_args, quiet=False) -> str: head_args = ["aws", "--profile", self.aws_profile] full_args = head_args + list(rest_args) if not quiet: print("$ ", end="") cprint(" ".join([shlex.quote(s) for s in full_args]), "cyan") ret = subprocess.check_output(full_args) return ret.decode().strip() def run_aws_cli_skip_err(self, *rest_args, quiet=False) -> Optional[str]: head_args = ["aws", "--profile", self.aws_profile] full_args = head_args + list(rest_args) if not quiet: print("$ ", end="") cprint(" ".join([shlex.quote(s) for s in full_args]), "cyan") try: ret = subprocess.check_output(full_args, stderr=subprocess.DEVNULL) except subprocess.CalledProcessError: cprint("SKIP", "red") return None return ret.decode().strip() def _sync_output_results(self): """CloudFormationのOutputの結果を辞書形式で得てインスタンス変数に格納。 """ ret = self.run_aws_cli( "cloudformation", "describe-stacks", "--stack-name", f"{self.prefix}-{self.stage}", ) rows = json.loads(ret)["Stacks"][0]["Outputs"] output_results_lst = [(row["OutputKey"], row["OutputValue"]) for row in rows] output_results_lst.sort(key=lambda x: x[0]) self._output_results = OrderedDict(output_results_lst) @property def output_results(self) -> OrderedDict: return self._output_results def pprint_output_results(self): """output_resultsを色付け&フォーマットして表示する """ if self._output_results is None: return for k, v in self._output_results.items(): cprint(f"{k}", "green", end="") print(f": {v}") return self._output_results PK!gtroposphere_ez/kit/__init__.pyfrom troposphere_ez.kit._cognito_simple_kit import CognitoSimpleKit from troposphere_ez.kit._dynamodb_simple_kit import DynamodbSimpleKit from troposphere_ez.kit._route53_simple_kit import Route53SimpleKit from troposphere_ez.kit._s3_simple_kit import S3SimpleKit from troposphere_ez.kit._spa_settings_kit import SpaSettingsKit from troposphere_ez.kit._sqs_simple_kit import SqsSimpleKit PK!j7!!)troposphere_ez/kit/_cognito_simple_kit.pyfrom typing import List, Union, Dict from troposphere import cognito, Ref, GetAtt from troposphere_ez import IStackManagerBase from troposphere_ez.utils import pascalyze_atom class CognitoSimpleKit(IStackManagerBase): def gen_email_valified_userpool( self, atom: str, allow_admin_create_user_only=True, require_lowercase=False, require_numbers=False, require_symbols=False, require_uppercase=False, ) -> cognito.UserPool: patom = pascalyze_atom(atom) up = cognito.UserPool(f"EmailValifiedUserPool{patom}") up.UserPoolName = self.prefix + "-" + atom + "-" + self.stage up.AdminCreateUserConfig = cognito.AdminCreateUserConfig( AllowAdminCreateUserOnly=allow_admin_create_user_only ) up.Policies = cognito.Policies( PasswordPolicy=cognito.PasswordPolicy( MinimumLength=8, RequireLowercase=False, RequireNumbers=False, RequireSymbols=False, RequireUppercase=False, ) ) up.UsernameAttributes = ["email"] up.AutoVerifiedAttributes = ["email"] return up def add_email_valified_userpool( self, atom: str, allow_admin_create_user_only=True, require_lowercase=False, require_numbers=False, require_symbols=False, require_uppercase=False, output=True, ) -> cognito.UserPool: """Eメールの自動検証がされるユーザプールを作成。 Args: allow_admin_create_user_only (bool): Trueにすると管理者のみがアカウント生成できる require_lowercase (bool) require_numbers (bool) require_symbols (bool) require_uppercase (bool) >>> _ = mgr.add_email_valified_userpool("foo-bar") >>> print(mgr.toyaml()) Outputs: EmailValifiedUserPoolFooBar: Value: !Ref 'EmailValifiedUserPoolFooBar' EmailValifiedUserPoolFooBarArn: Value: !GetAtt 'EmailValifiedUserPoolFooBar.Arn' EmailValifiedUserPoolFooBarProviderName: Value: !GetAtt 'EmailValifiedUserPoolFooBar.ProviderName' EmailValifiedUserPoolFooBarProviderURL: Value: !GetAtt 'EmailValifiedUserPoolFooBar.ProviderURL' Resources: EmailValifiedUserPoolFooBar: Properties: AdminCreateUserConfig: AllowAdminCreateUserOnly: 'true' AutoVerifiedAttributes: - email Policies: PasswordPolicy: MinimumLength: 8 RequireLowercase: 'false' RequireNumbers: 'false' RequireSymbols: 'false' RequireUppercase: 'false' UserPoolName: testtesttest-foo-bar-dev UsernameAttributes: - email Type: AWS::Cognito::UserPool """ up = self.gen_email_valified_userpool( atom, allow_admin_create_user_only, require_lowercase, require_numbers, require_symbols, require_uppercase, ) self.add_resource(up) if output: self.rec_to_output(up.title, Ref(up)) self.rec_to_output(up.title + "Arn", GetAtt(up, "Arn")) self.rec_to_output(up.title + "ProviderName", GetAtt(up, "ProviderName")) self.rec_to_output(up.title + "ProviderURL", GetAtt(up, "ProviderURL")) return up def gen_simple_userpool_client( self, atom: str, user_pool: cognito.UserPool ) -> cognito.UserPoolClient: patom = pascalyze_atom(atom) cli = cognito.UserPoolClient(f"SimpleUserPoolClient{patom}") # cli.ClientName = f"{self.config.basic['prefix']}-{self.config.stage}-{atom}" cli.UserPoolId = Ref(user_pool) cli.ExplicitAuthFlows = ["USER_PASSWORD_AUTH"] return cli def add_simple_userpool_client( self, atom: str, user_pool: cognito.UserPool, output=True ) -> cognito.UserPoolClient: """単純なUserPoolClientの作成。 >>> userpool = mgr.add_email_valified_userpool("account") >>> _ = mgr.add_simple_userpool_client("local", userpool) >>> print(mgr.toyaml()) Outputs: EmailValifiedUserPoolAccount: Value: !Ref 'EmailValifiedUserPoolAccount' EmailValifiedUserPoolAccountArn: Value: !GetAtt 'EmailValifiedUserPoolAccount.Arn' EmailValifiedUserPoolAccountProviderName: Value: !GetAtt 'EmailValifiedUserPoolAccount.ProviderName' EmailValifiedUserPoolAccountProviderURL: Value: !GetAtt 'EmailValifiedUserPoolAccount.ProviderURL' SimpleUserPoolClientLocal: Value: !Ref 'SimpleUserPoolClientLocal' Resources: EmailValifiedUserPoolAccount: Properties: AdminCreateUserConfig: AllowAdminCreateUserOnly: 'true' AutoVerifiedAttributes: - email Policies: PasswordPolicy: MinimumLength: 8 RequireLowercase: 'false' RequireNumbers: 'false' RequireSymbols: 'false' RequireUppercase: 'false' UserPoolName: testtesttest-account-dev UsernameAttributes: - email Type: AWS::Cognito::UserPool SimpleUserPoolClientLocal: Properties: ExplicitAuthFlows: - USER_PASSWORD_AUTH UserPoolId: !Ref 'EmailValifiedUserPoolAccount' Type: AWS::Cognito::UserPoolClient """ cli = self.gen_simple_userpool_client(atom, user_pool) self.add_resource(cli) if output: self.rec_to_output(cli.title, Ref(cli)) return cli def setup_identity_provider( self, up: Union[str, cognito.UserPool], provider_name: str, provider_type: str, provider_details: Dict[str, str], ): """Google/FacebookなどのIDプロバイダとUserPoolと連携する """ up_id = up if type(up) is str else self.output_results[up.title] provider_details_s = ",".join( [f"{k}='{v}'" for k, v in provider_details.items()] ) self.run_aws_cli( "cognito-idp", "create-identity-provider", "--user-pool-id", up_id, "--provider-name", provider_name, "--provider-type", provider_type, "--provider-details", provider_details_s, ) def setup_userpool_client( self, up: Union[str, cognito.UserPool], up_client: Union[str, cognito.UserPoolClient], signin_url: str, signout_url: str, providers=["COGNITO"], allowed_o_auth_flows=["code", "implicit"], allowd_o_auth_scopes=["openid", "email", "phone", "profile"], ): """CloudFormationだけでは設定不可能なUserPoolClientの設定をCLIで行う $ aws cognito-idp update-user-pool-client ... に対するラッパー関数 """ up_id = up if type(up) is str else self.output_results[up.title] up_client_id = ( up_client if type(up_client) is str else self.output_results[up_client.title] ) self.run_aws_cli_skip_err( "cognito-idp", "update-user-pool-client", "--user-pool-id", up_id, "--client-id", up_client_id, "--supported-identity-providers", *providers, "--callback-urls", f'["{signin_url}"]', "--logout-urls", f'["{signout_url}"]', f"--allowed-o-auth-flows", *allowed_o_auth_flows, "--allowed-o-auth-scopes", *allowd_o_auth_scopes, ) def setup_userpool_domain( self, up: Union[str, cognito.UserPool], domain_prefix: str ): """UserPoolに対する認証用WED画面のドメインURLを設定する """ up_id = up if type(up) is str else self.output_results[up.title] self.run_aws_cli_skip_err( "cognito-idp", "create-user-pool-domain", "--user-pool-id", up_id, "--domain", domain_prefix, ) PK!Z,U U *troposphere_ez/kit/_dynamodb_simple_kit.pyfrom troposphere import dynamodb, Ref, GetAtt from troposphere_ez import IStackManagerBase from troposphere_ez.utils import pascalyze_atom from typing import Dict class DynamodbSimpleKit(IStackManagerBase): def gen_simple_dynamodb_table( self, atom: str, attr_definitions: Dict[str, str], key_schema: Dict[str, str], rcu: int, wcu: int, ) -> dynamodb.Table: patom = pascalyze_atom(atom) dtbl = dynamodb.Table(title=f"SimpleDynamodbTable{patom}") dtbl.AttributeDefinitions = [ dynamodb.AttributeDefinition(AttributeName=name, AttributeType=atype) for (name, atype) in attr_definitions.items() ] dtbl.KeySchema = [ dynamodb.KeySchema(AttributeName=name, KeyType=ktype) for (name, ktype) in key_schema.items() ] dtbl.ProvisionedThroughput = dynamodb.ProvisionedThroughput( ReadCapacityUnits=rcu, WriteCapacityUnits=wcu ) return dtbl def add_simple_dynamodb_table( self, atom: str, attr_definitions: Dict[str, str], key_schema: Dict[str, str], rcu: int, wcu: int, output=True, ) -> dynamodb.Table: """DynamoTableを生成する >>> _ = mgr.add_simple_dynamodb_table("foo-bar", {"Symbol": "S"}, {"Symbol": "HASH"}, 3, 1) >>> print(mgr.toyaml()) Outputs: SimpleDynamodbTableFooBar: Value: !Ref 'SimpleDynamodbTableFooBar' SimpleDynamodbTableFooBarArn: Value: !GetAtt 'SimpleDynamodbTableFooBar.Arn' Resources: SimpleDynamodbTableFooBar: Properties: AttributeDefinitions: - AttributeName: Symbol AttributeType: S KeySchema: - AttributeName: Symbol KeyType: HASH ProvisionedThroughput: ReadCapacityUnits: 3 WriteCapacityUnits: 1 Type: AWS::DynamoDB::Table """ dtbl = self.gen_simple_dynamodb_table( atom, attr_definitions, key_schema, rcu, wcu ) self.add_resource(dtbl) if output: self.rec_to_output(dtbl.title, Ref(dtbl)) self.rec_to_output(dtbl.title + "Arn", GetAtt(dtbl, "Arn")) return dtbl PK!;;7 7 )troposphere_ez/kit/_route53_simple_kit.pyimport json from typing import Optional from troposphere import route53, Ref from troposphere_ez import IStackManagerBase from troposphere_ez.utils import pascalyze_atom class Route53SimpleKit(IStackManagerBase): def get_certificate_arn(self, domain_name: str) -> Optional[str]: """既に作ってあるSSL/TLS証明書のARNを得る。 """ ret = self.run_aws_cli( "acm", "list-certificates", "--region", "us-east-1", quiet=True ) rows = json.loads(ret)["CertificateSummaryList"] for row in rows: if row["DomainName"] == domain_name: return row["CertificateArn"] return None def gen_record_set_group_to_assign_cname( self, atom: str, hosted_zone_name: str, cname: str, original_name: str, ttl=300 ): patom = pascalyze_atom(atom) recg = route53.RecordSetGroup(f"RecordSetGroupToAssignCname{patom}") recg.HostedZoneName = hosted_zone_name recg.RecordSets = [ route53.RecordSet( Type="CNAME", Name=cname + ".", ResourceRecords=[original_name] ) ] return recg def add_record_set_group_to_assign_cname( self, atom: str, hosted_zone_name: str, cname: str, original_name, ttl=300, output=True, ): """CNAME割当を設定したレコードセットグループを追加する。 >>> _ = mgr.add_record_set_group_to_assign_cname("foo-bar", "example.com", "foo.example.com", "s3d6z3onlkl5.nantoka.com") >>> print(mgr.toyaml()) Outputs: RecordSetGroupToAssignCnameFooBar: Value: !Ref 'RecordSetGroupToAssignCnameFooBar' Resources: RecordSetGroupToAssignCnameFooBar: Properties: HostedZoneName: example.com RecordSets: - Name: foo.example.com. ResourceRecords: - s3d6z3onlkl5.nantoka.com Type: CNAME Type: AWS::Route53::RecordSetGroup """ recg = self.gen_record_set_group_to_assign_cname( atom, hosted_zone_name, cname, original_name ) self.add_resource(recg) if output: self.rec_to_output(recg.title, Ref(recg)) return recg PK!Y[ pp$troposphere_ez/kit/_s3_simple_kit.pyfrom typing import List, Tuple, Optional, Union from troposphere import s3, iam, cloudfront, Ref, GetAtt from troposphere_ez import IStackManagerBase from troposphere_ez.utils import pascalyze_atom, concat class S3SimpleKit(IStackManagerBase): def gen_secret_s3_bucket(self, atom: str) -> s3.Bucket: patom = pascalyze_atom(atom) bucket = s3.Bucket(f"SecretS3Bucket{patom}") bucket.AccessControl = "Private" bucket.PublicAccessBlockConfiguration = s3.PublicAccessBlockConfiguration( BlockPublicAcls=True, BlockPublicPolicy=True, IgnorePublicAcls=True, RestrictPublicBuckets=True, ) return bucket def add_secret_s3_bucket(self, atom: str, output=True) -> s3.Bucket: """非公開なs3バケットのインスタンスを作成しテンプレートへ追加 >>> _ = mgr.add_secret_s3_bucket("foo-bar") >>> print(mgr.toyaml()) Outputs: SecretS3BucketFooBar: Value: !Ref 'SecretS3BucketFooBar' SecretS3BucketFooBarArn: Value: !GetAtt 'SecretS3BucketFooBar.Arn' Resources: SecretS3BucketFooBar: Properties: AccessControl: Private PublicAccessBlockConfiguration: BlockPublicAcls: 'true' BlockPublicPolicy: 'true' IgnorePublicAcls: 'true' RestrictPublicBuckets: 'true' Type: AWS::S3::Bucket """ bucket = self.gen_secret_s3_bucket(atom) self.add_resource(bucket) if output: self.rec_to_output(bucket.title, Ref(bucket)) self.rec_to_output(bucket.title + "Arn", GetAtt(bucket, "Arn")) return bucket def gen_public_s3_bucket(self, atom: str) -> s3.Bucket: patom = pascalyze_atom(atom) bucket = s3.Bucket(f"PublicS3Bucket{patom}") bucket.AccessControl = "PublicRead" bucket.PublicAccessBlockConfiguration = s3.PublicAccessBlockConfiguration( BlockPublicAcls=False, BlockPublicPolicy=False, IgnorePublicAcls=False, RestrictPublicBuckets=False, ) return bucket def add_public_s3_bucket(self, atom: str, output=True) -> s3.Bucket: """パブリックに読み取り権限を与えたs3バケットのインスタンスを作成しテンプレートへ追加 >>> mgr.add_public_s3_bucket("foo-bar") and print(mgr.toyaml()) Outputs: PublicS3BucketFooBar: Value: !Ref 'PublicS3BucketFooBar' PublicS3BucketFooBarArn: Value: !GetAtt 'PublicS3BucketFooBar.Arn' Resources: PublicS3BucketFooBar: Properties: AccessControl: PublicRead PublicAccessBlockConfiguration: BlockPublicAcls: 'false' BlockPublicPolicy: 'false' IgnorePublicAcls: 'false' RestrictPublicBuckets: 'false' Type: AWS::S3::Bucket """ bucket = self.gen_public_s3_bucket(atom) self.add_resource(bucket) if output: self.rec_to_output(bucket.title, Ref(bucket)) self.rec_to_output(bucket.title + "Arn", GetAtt(bucket, "Arn")) return bucket def gen_policy_for_read_to_s3_bucket( self, atom: str, buckets: List[Union[Tuple[s3.Bucket, Optional[str]], s3.Bucket]], also_head=False, also_list=False, ) -> iam.Policy: patom = pascalyze_atom(atom) policy = iam.Policy(f"PolicyForReadToS3Bucket{patom}") res_on_first = [] for elm in buckets: if isinstance(elm, (list, tuple)): bucket, rest = elm[:2] else: bucket, rest = elm, None if rest: res_on_first.append(concat("arn:aws:s3:::", Ref(bucket), "/", rest)) else: res_on_first.append(concat("arn:aws:s3:::", Ref(bucket))) act_on_first = ["s3:GetObject"] if also_list: act_on_first.append("s3:ListBucket") statements = [ {"Effect": "Allow", "Action": act_on_first, "Resource": res_on_first} ] if also_head: statements.append( {"Effect": "Allow", "Action": "s3:HeadBucket", "Resource": "*"} ) policy.PolicyDocument = {"Version": "2012-10-17", "Statement": statements} return policy def gen_bucket_policy_for_cfoai_user( self, atom: str, bucket: s3.Bucket, cfoai: cloudfront.CloudFrontOriginAccessIdentity, ): patom = pascalyze_atom(atom) policy = s3.BucketPolicy(f"BucketPolicyForCfoaiUser{patom}") policy.Bucket = Ref(bucket) policy.PolicyDocument = { "Statement": [ { "Action": ["s3:GetObject"], "Effect": "Allow", "Resource": concat(GetAtt(bucket, "Arn"), "/*"), "Principal": {"CanonicalUser": GetAtt(cfoai, "S3CanonicalUserId")}, } ] } return policy def add_s3_bucket_policy_for_cfoai_user( self, atom: str, bucket: s3.Bucket, cfoai: cloudfront.CloudFrontOriginAccessIdentity, output=True, ): """cfoai(CloudFrontOriginAccessIdentity)経由のユーザに対して閲覧権を与える """ policy = self.gen_bucket_policy_for_cfoai_user(atom, bucket, cfoai) self.add_resource(policy) if output: self.rec_to_output(policy.title, Ref(bucket)) return policy PK!00'troposphere_ez/kit/_spa_settings_kit.pyfrom typing import Tuple, List from troposphere import s3, cloudfront, route53, Ref, GetAtt from troposphere_ez import IStackManagerBase from troposphere_ez.utils import pascalyze_atom, concat class SpaSettingsKit(IStackManagerBase): def gen_spa_settings_without_domain( self, atom: str, ttl: int ) -> Tuple[ s3.Bucket, cloudfront.CloudFrontOriginAccessIdentity, cloudfront.Distribution ]: patom = pascalyze_atom(atom) bucket = s3.Bucket(f"SpaWithoutDomainS3Bucket{patom}") bucket.AccessControl = "Private" bucket.PublicAccessBlockConfiguration = s3.PublicAccessBlockConfiguration( BlockPublicAcls=True, BlockPublicPolicy=True, IgnorePublicAcls=True, RestrictPublicBuckets=True, ) cfoai = cloudfront.CloudFrontOriginAccessIdentity( f"SpaWithoutDomainCloudFrontOriginAccessIdentity{patom}" ) cfoai.CloudFrontOriginAccessIdentityConfig = cloudfront.CloudFrontOriginAccessIdentityConfig( Comment=f"{self.aws_profile}-{self.stage}" ) cfdist = cloudfront.Distribution( f"SpaWithoutDomainCloudFrontDistribution{patom}" ) cfcfg = cloudfront.DistributionConfig() cfcfg.CustomErrorResponses = [ cloudfront.CustomErrorResponse( ErrorCachingMinTTL=0, ErrorCode=403, ResponseCode=200, ResponsePagePath="/", ) ] cfcfg.DefaultCacheBehavior = cloudfront.DefaultCacheBehavior( AllowedMethods=["HEAD", "GET"], TargetOriginId=concat("s3-", Ref(bucket)), ViewerProtocolPolicy="redirect-to-https", ForwardedValues=cloudfront.ForwardedValues(QueryString=True), DefaultTTL=ttl, ) cfcfg.DefaultRootObject = "index.html" cfcfg.Enabled = True cfcfg.Origins = [ cloudfront.Origin( DomainName=GetAtt(bucket, "DomainName"), Id=concat("s3-", Ref(bucket)), S3OriginConfig=cloudfront.S3OriginConfig( OriginAccessIdentity=concat( "origin-access-identity/cloudfront/", Ref(cfoai) ) ), ) ] cfdist.DistributionConfig = cfcfg return (bucket, cfoai, cfdist) def add_spa_settings_without_domain( self, atom: str, ttl: int, output=True ) -> Tuple[ s3.Bucket, cloudfront.CloudFrontOriginAccessIdentity, cloudfront.Distribution ]: """SPAを配信するためのスタック; ドメインなし版 >>> _ = mgr.add_spa_settings_without_domain("angular", 60) >>> print(mgr.toyaml()) Outputs: SpaWithoutDomainCloudFrontDistributionAngular: Value: !Ref 'SpaWithoutDomainCloudFrontDistributionAngular' SpaWithoutDomainCloudFrontOriginAccessIdentityAngular: Value: !Ref 'SpaWithoutDomainCloudFrontOriginAccessIdentityAngular' SpaWithoutDomainS3BucketAngular: Value: !Ref 'SpaWithoutDomainS3BucketAngular' Resources: SpaWithoutDomainCloudFrontDistributionAngular: Properties: DistributionConfig: CustomErrorResponses: - ErrorCachingMinTTL: 0 ErrorCode: 403 ResponseCode: 200 ResponsePagePath: / DefaultCacheBehavior: AllowedMethods: - HEAD - GET DefaultTTL: 60 ForwardedValues: QueryString: 'true' TargetOriginId: !Join - '' - - s3- - !Ref 'SpaWithoutDomainS3BucketAngular' ViewerProtocolPolicy: redirect-to-https DefaultRootObject: index.html Enabled: 'true' Origins: - DomainName: !GetAtt 'SpaWithoutDomainS3BucketAngular.DomainName' Id: !Join - '' - - s3- - !Ref 'SpaWithoutDomainS3BucketAngular' S3OriginConfig: OriginAccessIdentity: !Join - '' - - origin-access-identity/cloudfront/ - !Ref 'SpaWithoutDomainCloudFrontOriginAccessIdentityAngular' Type: AWS::CloudFront::Distribution SpaWithoutDomainCloudFrontOriginAccessIdentityAngular: Properties: CloudFrontOriginAccessIdentityConfig: Comment: default-dev Type: AWS::CloudFront::CloudFrontOriginAccessIdentity SpaWithoutDomainS3BucketAngular: Properties: AccessControl: Private PublicAccessBlockConfiguration: BlockPublicAcls: 'true' BlockPublicPolicy: 'true' IgnorePublicAcls: 'true' RestrictPublicBuckets: 'true' Type: AWS::S3::Bucket """ (bucket, cfoai, cfdist) = self.gen_spa_settings_without_domain(atom, ttl) self.add_resource(bucket) self.add_resource(cfoai) self.add_resource(cfdist) if output: self.rec_to_output(bucket.title, Ref(bucket)) self.rec_to_output(cfoai.title, Ref(cfoai)) self.rec_to_output(cfdist.title, Ref(cfdist)) return (bucket, cfoai, cfdist) def gen_spa_settings( self, atom: str, ttl: int, hosted_zone_name: str, domain_name: str, certificate_arn: str, ) -> Tuple[ s3.Bucket, cloudfront.CloudFrontOriginAccessIdentity, cloudfront.Distribution, route53.RecordSetGroup, ]: patom = pascalyze_atom(atom) bucket = s3.Bucket(f"SpaS3Bucket{patom}") bucket.AccessControl = "Private" bucket.PublicAccessBlockConfiguration = s3.PublicAccessBlockConfiguration( BlockPublicAcls=True, BlockPublicPolicy=True, IgnorePublicAcls=True, RestrictPublicBuckets=True, ) cfoai = cloudfront.CloudFrontOriginAccessIdentity( f"SpaCloudFrontOriginAccessIdentity{patom}" ) cfoai.CloudFrontOriginAccessIdentityConfig = cloudfront.CloudFrontOriginAccessIdentityConfig( Comment=f"{self.aws_profile}-{self.stage}" ) cfdist = cloudfront.Distribution(f"SpaCloudFrontDistribution{patom}") cfcfg = cloudfront.DistributionConfig() cfcfg.CustomErrorResponses = [ cloudfront.CustomErrorResponse( ErrorCachingMinTTL=0, ErrorCode=403, ResponseCode=200, ResponsePagePath="/", ) ] cfcfg.DefaultCacheBehavior = cloudfront.DefaultCacheBehavior( AllowedMethods=["HEAD", "GET"], TargetOriginId=concat("s3-", Ref(bucket)), ViewerProtocolPolicy="redirect-to-https", ForwardedValues=cloudfront.ForwardedValues(QueryString=True), DefaultTTL=ttl, ) cfcfg.DefaultRootObject = "index.html" cfcfg.Enabled = True cfcfg.Origins = [ cloudfront.Origin( DomainName=GetAtt(bucket, "DomainName"), Id=concat("s3-", Ref(bucket)), S3OriginConfig=cloudfront.S3OriginConfig( OriginAccessIdentity=concat( "origin-access-identity/cloudfront/", Ref(cfoai) ) ), ) ] cfcfg.ViewerCertificate = cloudfront.ViewerCertificate( AcmCertificateArn=certificate_arn, SslSupportMethod="sni-only" ) cfdist.DistributionConfig = cfcfg recg = route53.RecordSetGroup( f"SpaRecordSetGroup{patom}", HostedZoneName=f"{hosted_zone_name}.", RecordSets=[ route53.RecordSet( Type="CNAME", Name=f"{domain_name}.", ResourceRecords=[GetAtt(cfdist, "DomainName")], TTL=300, ) ], ) return (bucket, cfoai, cfdist, recg) def add_spa_settings( self, atom: str, ttl: int, hosted_zone_name: str, domain_name: str, certificate_arn: str, output=True, ) -> Tuple[ s3.Bucket, cloudfront.CloudFrontOriginAccessIdentity, cloudfront.Distribution, route53.RecordSetGroup, ]: """SPAを配信するためのスタック; ドメインなし版 >>> _ = mgr.add_spa_settings("angular", 60, "example.com", "foo.example.com", "s3d6z3onlkl5.nantoka.com", "arn:aws:acm:us-east-1:XXX:certificate/XXX") >>> print(mgr.toyaml()) Outputs: SpaCloudFrontDistributionAngular: Value: !Ref 'SpaCloudFrontDistributionAngular' SpaCloudFrontOriginAccessIdentityAngular: Value: !Ref 'SpaCloudFrontOriginAccessIdentityAngular' SpaRecordSetGroupAngular: Value: !Ref 'SpaRecordSetGroupAngular' SpaS3BucketAngular: Value: !Ref 'SpaS3BucketAngular' Resources: SpaCloudFrontDistributionAngular: Properties: DistributionConfig: CustomErrorResponses: - ErrorCachingMinTTL: 0 ErrorCode: 403 ResponseCode: 200 ResponsePagePath: / DefaultCacheBehavior: AllowedMethods: - HEAD - GET DefaultTTL: 60 ForwardedValues: QueryString: 'true' TargetOriginId: !Join - '' - - s3- - !Ref 'SpaS3BucketAngular' ViewerProtocolPolicy: redirect-to-https DefaultRootObject: index.html Enabled: 'true' Origins: - DomainName: !GetAtt 'SpaS3BucketAngular.DomainName' Id: !Join - '' - - s3- - !Ref 'SpaS3BucketAngular' S3OriginConfig: OriginAccessIdentity: !Join - '' - - origin-access-identity/cloudfront/ - !Ref 'SpaCloudFrontOriginAccessIdentityAngular' ViewerCertificate: AcmCertificateArn: s3d6z3onlkl5.nantoka.com SslSupportMethod: sni-only Type: AWS::CloudFront::Distribution SpaCloudFrontOriginAccessIdentityAngular: Properties: CloudFrontOriginAccessIdentityConfig: Comment: default-dev Type: AWS::CloudFront::CloudFrontOriginAccessIdentity SpaRecordSetGroupAngular: Properties: HostedZoneName: example.com. RecordSets: - Name: foo.example.com. ResourceRecords: - !GetAtt 'SpaCloudFrontDistributionAngular.DomainName' TTL: 300 Type: CNAME Type: AWS::Route53::RecordSetGroup SpaS3BucketAngular: Properties: AccessControl: Private PublicAccessBlockConfiguration: BlockPublicAcls: 'true' BlockPublicPolicy: 'true' IgnorePublicAcls: 'true' RestrictPublicBuckets: 'true' Type: AWS::S3::Bucket """ (bucket, cfoai, cfdist, recg) = self.gen_spa_settings( atom, ttl, hosted_zone_name, domain_name, certificate_arn ) self.add_resource(bucket) self.add_resource(cfoai) self.add_resource(cfdist) self.add_resource(recg) if output: self.rec_to_output(bucket.title, Ref(bucket)) self.rec_to_output(cfoai.title, Ref(cfoai)) self.rec_to_output(cfdist.title, Ref(cfdist)) self.rec_to_output(recg.title, Ref(recg)) return (bucket, cfoai, cfdist, recg) PK![oo%troposphere_ez/kit/_sqs_simple_kit.pyfrom troposphere import sqs, Ref, GetAtt from troposphere_ez import IStackManagerBase from troposphere_ez.utils import pascalyze_atom class SqsSimpleKit(IStackManagerBase): def add_fifo_sqs_queue(self, atom: str, output=True) -> sqs.Queue: patom = pascalyze_atom(atom) que = sqs.Queue(title=f"FifoSqsQueue{patom}") que.FifoQueue = True que.ContentBasedDeduplication = True self.add_resource(que) if output: self.rec_to_output(que.title, Ref(que)) self.rec_to_output(que.title + "Arn", GetAtt(que, "Arn")) self.rec_to_output(que.title + "QueueName", GetAtt(que, "QueueName")) return que def add_simple_standard_sqs_queue(self, atom: str, output=True) -> sqs.Queue: patom = pascalyze_atom(atom) que = sqs.Queue(title=f"StandardSqsQueue{patom}") self.add_resource(que) if output: self.rec_to_output(que.title, Ref(que)) self.rec_to_output(que.title + "Arn", GetAtt(que, "Arn")) self.rec_to_output(que.title + "QueueName", GetAtt(que, "QueueName")) return que PK!*tttroposphere_ez/utils.pyimport re from stringcase import pascalcase from troposphere import Join def concat(*args): return Join("", args) def pascalyze_atom(atom: str) -> str: """hoge-fooのような名前をHogeFooへ置き換える """ if not re.search(r"^[a-z][a-z-]*$", atom) or atom[-1] == "-": raise ValueError() return pascalcase(atom.replace("-", "_")) PK!HnHTU$troposphere_ez-0.1.3.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HS E'troposphere_ez-0.1.3.dist-info/METADATAO0Wܣ&[dž"ihDDAK9um%Ƙֻݡ[E֑8,B#oMm\b7cGFka[ k^iт,sMT3AIdM(p#IXyMƤEWy_;yl4:Ua+ӌ$V.h;d "W`ywڳ70i|&4`6PiIdQu=7$^