#!/usr/bin/env python
"""Module with static website bucket and CloudFront distribution."""
from __future__ import print_function
import hashlib
import logging
import os
from typing import Any, Dict, List, Union # pylint: disable=unused-import
import awacs.awslambda
import awacs.iam
import awacs.logs
import awacs.s3
import awacs.states
import awacs.sts
from awacs.aws import Action, Allow, Policy, PolicyDocument, Principal, Statement
from awacs.helpers.trust import make_simple_assume_policy
from troposphere import (
AccountId,
Join,
NoValue,
Output,
Partition,
Region,
StackName,
awslambda,
cloudfront,
iam,
s3,
)
from runway.cfngin.blueprints.base import Blueprint
from runway.cfngin.context import Context
LOGGER = logging.getLogger("runway")
IAM_ARN_PREFIX = "arn:aws:iam::aws:policy/service-role/"
[docs]class StaticSite(Blueprint): # pylint: disable=too-few-public-methods
"""CFNgin blueprint for creating S3 bucket and CloudFront distribution."""
VARIABLES = {
"AcmCertificateArn": {
"type": str,
"default": "",
"description": "(Optional) Cert ARN for site",
},
"Aliases": {
"type": list,
"default": [],
"description": "(Optional) Domain aliases the " "distribution",
},
"DisableCloudFront": {
"type": bool,
"default": False,
"description": "Whether to disable CF",
},
"LogBucketName": {
"type": str,
"default": "",
"description": "S3 bucket for CF logs",
},
"PriceClass": {
"type": str,
"default": "PriceClass_100", # US/Europe
"description": "CF price class for the distribution.",
},
"RewriteDirectoryIndex": {
"type": str,
"default": "",
"description": "(Optional) File name to "
"append to directory "
"requests.",
},
"RoleBoundaryArn": {
"type": str,
"default": "",
"description": "(Optional) IAM Role permissions "
"boundary applied to any created "
"roles.",
},
"WAFWebACL": {
"type": str,
"default": "",
"description": "(Optional) WAF id to associate with the " "distribution.",
},
"custom_error_responses": {
"type": list,
"default": [],
"description": "(Optional) Custom error " "responses.",
},
"lambda_function_associations": {
"type": list,
"default": [],
"description": "(Optional) Lambda " "function " "associations.",
},
}
@property
def aliases_specified(self):
# type: () -> bool
"""Aliases are specified conditional."""
return self.get_variables()["Aliases"] != [""]
@property
def cf_enabled(self):
# type: () -> bool
"""CloudFront enabled conditional."""
return not self.get_variables().get("DisableCloudFront", False)
@property
def acm_certificate_specified(self):
# type: () -> bool
"""ACM Certification specified conditional."""
return self.get_variables()["AcmCertificateArn"] != ""
@property
def cf_logging_enabled(self):
# type: () -> bool
"""CloudFront Logging specified conditional."""
return self.get_variables()["LogBucketName"] != ""
@property
def directory_index_specified(self):
# type: () -> bool
"""Directory Index specified conditional."""
return self.get_variables()["RewriteDirectoryIndex"] != ""
@property
def role_boundary_specified(self):
# type: () -> bool
"""IAM Role Boundary specified conditional."""
return self.get_variables()["RoleBoundaryArn"] != ""
@property
def waf_name_specified(self):
# type: () -> bool
"""WAF name specified conditional."""
return self.get_variables()["WAFWebACL"] != ""
[docs] def create_template(self):
# type: () -> None
"""Create template (main function called by CFNgin)."""
self.template.set_version("2010-09-09")
self.template.set_description("Static Website - Bucket and Distribution")
# Resources
bucket = self.add_bucket()
if self.cf_enabled:
oai = self.add_origin_access_identity()
bucket_policy = self.add_cloudfront_bucket_policy(bucket, oai)
lambda_function_associations = self.get_lambda_associations()
if self.directory_index_specified:
index_rewrite = self._get_index_rewrite_role_function_and_version()
lambda_function_associations = self.get_directory_index_lambda_association(
lambda_function_associations, index_rewrite["version"]
)
distribution_options = self.get_cloudfront_distribution_options(
bucket, oai, lambda_function_associations
)
distribution = self.add_cloudfront_distribution( # noqa pylint: disable=unused-variable
bucket_policy, distribution_options
)
else:
self.add_bucket_policy(bucket)
[docs] def get_lambda_associations(self):
# type: () -> List[cloudfront.LambdaFunctionAssociation]
"""Retrieve any lambda associations from the instance variables.
Return:
List of Lambda Function association variables
"""
variables = self.get_variables()
# If custom associations defined, use them
if variables["lambda_function_associations"]:
return [
cloudfront.LambdaFunctionAssociation(
EventType=x["type"], LambdaFunctionARN=x["arn"]
)
for x in variables["lambda_function_associations"]
]
return []
[docs] def get_directory_index_lambda_association( # pylint: disable=no-self-use
self,
lambda_associations, # type: List[cloudfront.LambdaFunctionAssociation]
directory_index_rewrite_version, # type: awslambda.Version
):
# type: (...) -> List[cloudfront.LambdaFunctionAssociation]
"""Retrieve the directory index lambda associations with the added rewriter.
Args:
lambda_associations: The lambda associations.
directory_index_rewrite_version: The directory index rewrite version.
"""
lambda_associations.append(
cloudfront.LambdaFunctionAssociation(
EventType="origin-request",
LambdaFunctionARN=directory_index_rewrite_version.ref(),
)
)
return lambda_associations
[docs] def get_cloudfront_distribution_options(
self,
bucket, # type: s3.Bucket
oai, # type: cloudfront.CloudFrontOriginAccessIdentity
lambda_function_associations, # type: List[cloudfront.LambdaFunctionAssociation]
):
# type: (...) -> Dict[str, Any]
"""Retrieve the options for our CloudFront distribution.
Args:
bucket: The bucket resource
oai: The origin access identity resource.
lambda_function_associations: List of Lambda Function associations.
Return:
The CloudFront Distribution Options.
"""
variables = self.get_variables()
if os.getenv("AWS_REGION") == "us-east-1":
# use global endpoint for us-east-1
origin = Join(".", [bucket.ref(), "s3.amazonaws.com"])
else:
# use reginal endpoint to avoid "temporary" redirect that can last over an hour
# https://forums.aws.amazon.com/message.jspa?messageID=677452
origin = Join(".", [bucket.ref(), "s3", Region, "amazonaws.com"])
return {
"Aliases": self.add_aliases(),
"Origins": [
cloudfront.Origin(
DomainName=origin,
S3OriginConfig=cloudfront.S3OriginConfig(
OriginAccessIdentity=Join(
"", ["origin-access-identity/cloudfront/", oai.ref()]
)
),
Id="S3Origin",
)
],
"DefaultCacheBehavior": cloudfront.DefaultCacheBehavior(
AllowedMethods=["GET", "HEAD"],
Compress=False,
DefaultTTL="86400",
ForwardedValues=cloudfront.ForwardedValues(
Cookies=cloudfront.Cookies(Forward="none"), QueryString=False,
),
LambdaFunctionAssociations=lambda_function_associations,
TargetOriginId="S3Origin",
ViewerProtocolPolicy="redirect-to-https",
),
"DefaultRootObject": "index.html",
"Logging": self.add_logging_bucket(),
"PriceClass": variables["PriceClass"],
"CustomErrorResponses": [
cloudfront.CustomErrorResponse(
ErrorCode=response["ErrorCode"],
ResponseCode=response["ResponseCode"],
ResponsePagePath=response["ResponsePagePath"],
)
for response in variables["custom_error_responses"]
],
"Enabled": True,
"WebACLId": self.add_web_acl(),
"ViewerCertificate": self.add_acm_cert(),
}
[docs] def add_aliases(self):
# type: () -> Union[List[str], NoValue]
"""Add aliases."""
if self.aliases_specified:
return self.get_variables()["Aliases"]
return NoValue
[docs] def add_web_acl(self):
# type: () -> Union[str, NoValue]
"""Add Web ACL."""
if self.waf_name_specified:
return self.get_variables()["WAFWebACL"]
return NoValue
[docs] def add_logging_bucket(self):
# type: () -> Union[cloudfront.Logging, NoValue]
"""Add Logging Bucket."""
if self.cf_logging_enabled:
return cloudfront.Logging(
Bucket=Join(
".", [self.get_variables()["LogBucketName"], "s3.amazonaws.com"]
)
)
return NoValue
[docs] def add_acm_cert(self):
# type: () -> Union[cloudfront.ViewerCertificate, NoValue]
"""Add ACM cert."""
if self.acm_certificate_specified:
return cloudfront.ViewerCertificate(
AcmCertificateArn=self.get_variables()["AcmCertificateArn"],
SslSupportMethod="sni-only",
)
return NoValue
[docs] def add_origin_access_identity(self):
# type: () -> cloudfront.CloudFrontOriginAccessIdentity
"""Add the origin access identity resource to the template.
Returns:
The OAI resource
"""
return self.template.add_resource(
cloudfront.CloudFrontOriginAccessIdentity(
"OAI",
CloudFrontOriginAccessIdentityConfig=cloudfront.CloudFrontOriginAccessIdentityConfig( # noqa
Comment="CF access to website"
),
)
)
[docs] def add_bucket_policy(self, bucket):
# type: (s3.Bucket) -> s3.BucketPolicy
"""Add a policy to the bucket if CloudFront is disabled. Ensure PublicRead.
Args:
bucket: The bucket resource to place the policy.
Returns:
The Bucket Policy Resource.
"""
return self.template.add_resource(
s3.BucketPolicy(
"BucketPolicy",
Bucket=bucket.ref(),
PolicyDocument=Policy(
Version="2012-10-17",
Statement=[
Statement(
Effect=Allow,
Principal=Principal("*"),
Action=[Action("s3", "getObject")],
Resource=[Join("", [bucket.get_att("Arn"), "/*"])],
)
],
),
)
)
[docs] def add_bucket(self):
# type: () -> s3.Bucket
"""Add the bucket resource along with an output of it's name / website url.
Returns:
The bucket resource.
"""
bucket = self.template.add_resource(
s3.Bucket(
"Bucket",
AccessControl=(s3.Private if self.cf_enabled else s3.PublicRead),
LifecycleConfiguration=s3.LifecycleConfiguration(
Rules=[
s3.LifecycleRule(
NoncurrentVersionExpirationInDays=90, Status="Enabled"
)
]
),
VersioningConfiguration=s3.VersioningConfiguration(Status="Enabled"),
)
)
self.template.add_output(
Output(
"BucketName", Description="Name of website bucket", Value=bucket.ref()
)
)
if not self.cf_enabled:
# bucket cannot be configured with WebsiteConfiguration when using OAI S3Origin
bucket.WebsiteConfiguration = s3.WebsiteConfiguration(
IndexDocument="index.html", ErrorDocument="error.html"
)
self.template.add_output(
Output(
"BucketWebsiteURL",
Description="URL of the bucket website",
Value=bucket.get_att("WebsiteURL"),
)
)
return bucket
[docs] def add_cloudfront_bucket_policy(self, bucket, oai):
# type (s3.Bucket, cloudfront.CloudFrontOriginAccessIdentity) -> s3.BucketPolicy
"""Given a bucket and oai resource add cloudfront access to the bucket.
Keyword Args:
bucket: A bucket resource.
oai: An Origin Access Identity resource.
Return:
The CloudFront Bucket access resource.
"""
return self.template.add_resource(
s3.BucketPolicy(
"AllowCFAccess",
Bucket=bucket.ref(),
PolicyDocument=PolicyDocument(
Version="2012-10-17",
Statement=self._get_cloudfront_bucket_policy_statements(
bucket, oai
),
),
)
)
[docs] def add_lambda_execution_role(
self,
name="LambdaExecutionRole", # type: str
function_name="", # type: str
): # noqa: E124
# type: (...) -> iam.Role
"""Create the Lambda@Edge execution role.
Args:
name: Name for the Lambda Execution Role.
function_name: Name of the Lambda Function the Role will be
attached to.
"""
variables = self.get_variables()
lambda_resource = Join(
"",
[
"arn:",
Partition,
":logs:*:",
AccountId,
":log-group:/aws/lambda/",
StackName,
"-%s-*" % function_name,
],
)
edge_resource = Join(
"",
[
"arn:",
Partition,
":logs:*:",
AccountId,
":log-group:/aws/lambda/*.",
StackName,
"-%s-*" % function_name,
],
)
return self.template.add_resource(
iam.Role(
name,
AssumeRolePolicyDocument=make_simple_assume_policy(
"lambda.amazonaws.com", "edgelambda.amazonaws.com"
),
PermissionsBoundary=(
variables["RoleBoundaryArn"]
if self.role_boundary_specified
else NoValue
),
Policies=[
iam.Policy(
PolicyName="LambdaLogCreation",
PolicyDocument=PolicyDocument(
Version="2012-10-17",
Statement=[
Statement(
Action=[
awacs.logs.CreateLogGroup,
awacs.logs.CreateLogStream,
awacs.logs.PutLogEvents,
],
Effect=Allow,
Resource=[lambda_resource, edge_resource],
)
],
),
),
],
)
)
[docs] def add_cloudfront_directory_index_rewrite(self, role):
# type: (iam.Role) -> awslambda.Function
"""Add an index CloudFront directory index rewrite lambda function to the template.
Keyword Args:
role: The index rewrite role resource.
Return:
The CloudFront directory index rewrite lambda function resource.
"""
variables = self.get_variables()
code_str = ""
path = os.path.join(
os.path.dirname(__file__),
"templates/cf_directory_index_rewrite.template.js",
)
with open(path) as file_:
code_str = file_.read().replace(
"{{RewriteDirectoryIndex}}", variables["RewriteDirectoryIndex"]
)
function = self.template.add_resource(
awslambda.Function(
"CFDirectoryIndexRewrite",
Code=awslambda.Code(ZipFile=code_str),
DeletionPolicy="Retain",
Description="Rewrites CF directory HTTP requests to default page",
Handler="index.handler",
Role=role.get_att("Arn"),
Runtime="nodejs10.x",
)
)
self.template.add_output(
Output(
"LambdaCFDirectoryIndexRewriteArn",
Description="Directory Index Rewrite Function Arn",
Value=function.get_att("Arn"),
)
)
return function
[docs] def add_cloudfront_directory_index_rewrite_version(self, directory_index_rewrite):
# type: (awslambda.Function) -> awslambda.Version
"""Add a specific version to the directory index rewrite lambda.
Args:
directory_index_rewrite (dict): The directory index rewrite lambda resource.
Return:
dict: The CloudFront directory index rewrite version.
"""
code_hash = hashlib.md5(
str(
directory_index_rewrite.properties["Code"].properties["ZipFile"]
).encode()
).hexdigest()
return self.template.add_resource(
awslambda.Version(
"CFDirectoryIndexRewriteVer" + code_hash,
FunctionName=directory_index_rewrite.ref(),
)
)
[docs] def add_cloudfront_distribution(
self, bucket_policy, cloudfront_distribution_options
):
# type: (s3.BucketPolicy, Dict[str, Any]) -> cloudfront.Distribution
"""Add the CloudFront distribution to the template / output the id and domain name.
Args:
bucket_policy (dict): Bucket policy to allow CloudFront access.
cloudfront_distribution_options (dict): The distribution options.
Return:
dict: The CloudFront Distribution resource
"""
distribution = self.template.add_resource(
cloudfront.Distribution(
"CFDistribution",
DependsOn=bucket_policy.title,
DistributionConfig=cloudfront.DistributionConfig(
**cloudfront_distribution_options
),
)
)
self.template.add_output(
Output(
"CFDistributionId",
Description="CloudFront distribution ID",
Value=distribution.ref(),
)
)
self.template.add_output(
Output(
"CFDistributionDomainName",
Description="CloudFront distribution domain name",
Value=distribution.get_att("DomainName"),
)
)
return distribution
def _get_cloudfront_bucket_policy_statements( # pylint: disable=no-self-use
self, bucket, oai
):
return [
Statement(
Action=[awacs.s3.GetObject],
Effect=Allow,
# S3CanonicalUserId is translated to the ARN when AWS renders this
Principal=Principal("CanonicalUser", oai.get_att("S3CanonicalUserId")),
Resource=[Join("", [bucket.get_att("Arn"), "/*"])],
)
]
def _get_index_rewrite_role_function_and_version(self):
res = {}
res["role"] = self.add_lambda_execution_role(
"CFDirectoryIndexRewriteRole", "CFDirectoryIndexRewrite"
)
res["function"] = self.add_cloudfront_directory_index_rewrite(res["role"])
res["version"] = self.add_cloudfront_directory_index_rewrite_version(
res["function"]
)
return res
# Helper section to enable easy blueprint -> template generation
# (just run `python <thisfile>` to output the json)
if __name__ == "__main__":
print(StaticSite("test", Context({"namespace": "test"}), None).to_json())