Encrypted storage¶

Four resources show up in nearly every data-handling stack, each with a handful of settings that should be the same every time: an encryption key, a private bucket, a table, and a container repository. These patterns set those defaults once.

import re
from collections.abc import Mapping, Sequence

from cfn_typed_sumo.types import dynamodb, ecr, kms, s3, ssm

from cfn_typed import Expr, Resources, Tag, allow, deny, get_att, join, ref, sub
from cfn_typed.dynamodb import Index, Key, keys
from cfn_typed.permissions import Ecr
from cfn_typed.policy import ANY, GlobalKeys, aws, bool_, string_not_like

An encryption key¶

def encryption_key(name: str, *, alias: str, description: str = "", tags: list[Expr[Tag]] | None = None) -> Resources:
    """A KMS key with rotation and an alias: logical IDs `{name}Key` and `{name}KeyAlias`.

    The key policy only enables IAM: access is then granted with ordinary identity policies. Both resources
    are retained when the stack is deleted, because deleting a key makes everything encrypted with it unreadable.
    """
    key = f"{name}Key"
    props: kms.KeyProps = {
        "Description": description or f"{name} encryption key",
        "EnableKeyRotation": True,
        "KeyPolicy": {
            "Version": "2012-10-17",
            "Statement": [
                # in a key policy Resource "*" means the key itself; naming its ARN would be a circular reference
                allow("kms:*", "*", sid="EnableIamPolicies", principal=aws(sub("arn:${AWS::Partition}:iam::${AWS::AccountId}:root"))),
            ],
        },
    }
    if tags:
        props["Tags"] = tags
    return {
        key: kms.key(deletion_policy="Retain", update_replace_policy="Retain", **props),
        f"{name}KeyAlias": kms.alias(
            AliasName=f"alias/{alias}", TargetKeyId=ref(key), deletion_policy="Retain", update_replace_policy="Retain",
        ),
    }

The key policy only enables IAM. Access is then granted with ordinary identity policies, for example with grants, instead of listing principals inside the key.
Resource is "*" in the key policy. In a key policy that means the key itself. Using the key's own ARN there is a circular reference, which is why the policy cannot name it.
Both resources are retained. Deleting a key makes everything encrypted with it unreadable, so a stack delete should never take it along.

A private bucket and its policy¶

def secure_bucket(
    name: str,
    *,
    key_arn: Expr[str],
    allowed_principals: Sequence[Expr[str]] | None = None,
    archive_after_days: int | None = None,
    tags: list[Expr[Tag]] | None = None,
) -> Resources:
    """A private, encrypted bucket and its policy: logical IDs `{name}Bucket` and `{name}BucketPolicy`.

    Public access is blocked, objects are encrypted with `key_arn`, and the bucket is retained on delete. The
    policy refuses plain HTTP. With `allowed_principals` it also refuses everyone else, so include the role that
    deploys the stack, or you will lock yourself out of the bucket.
    """
    bucket = f"{name}Bucket"
    bucket_arn = get_att(bucket, "Arn")
    both: list[Expr[str]] = [bucket_arn, join("", [bucket_arn, "/*"])]
    statements = [
        deny("s3:*", both, sid="DenyInsecureTransport", principal=ANY, condition=bool_(GlobalKeys.SecureTransport, False)),
    ]
    if allowed_principals:
        statements.append(
            deny("s3:*", both, sid="DenyEveryoneElse", principal=ANY,
                 condition=string_not_like(GlobalKeys.PrincipalArn, list(allowed_principals))),
        )
    props: s3.BucketProps = {
        "PublicAccessBlockConfiguration": {
            "BlockPublicAcls": True, "BlockPublicPolicy": True, "IgnorePublicAcls": True, "RestrictPublicBuckets": True,
        },
        "BucketEncryption": {"ServerSideEncryptionConfiguration": [{
            "ServerSideEncryptionByDefault": {"SSEAlgorithm": "aws:kms", "KMSMasterKeyID": key_arn},
            "BucketKeyEnabled": True,
        }]},
    }
    if archive_after_days is not None:
        props["LifecycleConfiguration"] = {"Rules": [{
            "Id": "archive", "Status": "Enabled",
            "Transitions": [{"StorageClass": "GLACIER", "TransitionInDays": archive_after_days}],
        }]}
    if tags:
        props["Tags"] = tags
    return {
        bucket: s3.bucket(deletion_policy="Retain", update_replace_policy="Retain", **props),
        f"{name}BucketPolicy": s3.bucket_policy(
            Bucket=ref(bucket), PolicyDocument={"Version": "2012-10-17", "Statement": statements},
        ),
    }

No bucket name. CloudFormation picks a unique one, so a replacement never collides with the old bucket. The next page shows how other stacks find it.
Two policy statements, both Deny. A deny for plain HTTP, and with allowed_principals a deny for everyone not listed. Deny statements win over any allow, so they are a backstop, not a grant.
The lockout risk is real. With allowed_principals, a principal that is not on the list cannot use the bucket, including the role that deploys the stack. The docstring says so, and the example stack takes the deploy role as a parameter for exactly this reason. Wildcards work: role/team-* matches a family of roles.

A table¶

def dynamodb_table(
    pk: Key,
    sk: Key | None = None,
    *,
    gsis: Sequence[Index] = (),
    lsis: Sequence[Index] = (),
    deletion_protection: bool = True,
    tags: list[Expr[Tag]] | None = None,
) -> dynamodb.Table:
    """An on-demand table with point-in-time recovery. Keys are `"name"` (a string) or `("name", "N")`."""
    return dynamodb.table(
        **keys(pk, sk, gsis=gsis, lsis=lsis),
        BillingMode="PAY_PER_REQUEST",
        PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
        DeletionProtectionEnabled=deletion_protection,
        **({"Tags": tags} if tags else {}),
    )

On-demand billing and point-in-time recovery are the defaults you almost always want, and deletion protection is on unless you turn it off. The pattern returns the table itself, so you choose its logical ID.

The key properties come from cfn_typed.dynamodb.keys. It is entirely optional. dynamodb.table(...) takes AttributeDefinitions, KeySchema and the indexes as ordinary typed properties, and writing them by hand is fine, and often clearer for a simple table. keys only saves the bookkeeping, mostly once a table has indexes:

table = dynamodb_table("pk", ("ts", "N"), gsis=[gsi("by-user", "user", projection=["name"])])

A key is "name" or ("name", "N"). A bare name is a string attribute.
AttributeDefinitions is derived. CloudFormation wants every key of the table and of each index declared exactly once. keys collects them, and refuses one name with two types.
The result is a plain dict. Spread it into the factory, dynamodb.table(**keys(...), BillingMode=...), and ty still checks everything you pass next to it. The keys themselves are correct by construction, so they are not typed against the generated shapes: that keeps this module free of generated types, with nothing to drift. The arguments are typed, though: an attribute type is "S", "N" or "B", and a projection is "ALL", "KEYS_ONLY" or a list of names.
lsi("name", sk) is a local index: it takes the table's hash key.

A container repository¶

def ecr_repository(
    name: str,
    *,
    pull_principals: Sequence[Expr[str]] = (),
    tags: list[Expr[Tag]] | None = None,
) -> ecr.Repository:
    """A repository with immutable tags and scan-on-push, that the given principals may pull from."""
    props: ecr.RepositoryProps = {
        "RepositoryName": name,
        "ImageTagMutability": "IMMUTABLE",
        "ImageScanningConfiguration": {"ScanOnPush": True},
    }
    if pull_principals:
        props["RepositoryPolicyText"] = {
            "Version": "2012-10-17",
            "Statement": [
                # no Resource: a repository policy applies to the repository it is attached to
                allow([Ecr.GetDownloadUrlForLayer, Ecr.BatchGetImage, Ecr.BatchCheckLayerAvailability],
                      sid="AllowPull", principal=aws(*pull_principals)),
            ],
        }
    if tags:
        props["Tags"] = tags
    return ecr.repository(**props)

Immutable tags and scan-on-push, and a pull policy only when you name principals. The pull actions are written with the permission names (Ecr.BatchGetImage and so on), so a typo in one is an error.

Resource policies¶

The bucket, key and repository policies above are three different shapes of resource policy. They are built with the policy helpers, which is why their shapes differ in the table:

Policy	`Principal`	`Resource`
bucket policy (its own resource)	`ANY`, narrowed by conditions	the bucket and its objects
key policy (inside the key)	an account root	`"*"`, meaning the key itself
repository policy	the principals that may pull	none: it applies to its repository

The two bucket statements are deny guardrails with conditions, and the key and repository policies are allow statements. Grants can produce the allow side too, with a principal.