如何为CDK函数编写辅助/装饰器函数复用S3公共访问阻断配置

该需求完全可落地实现，以下是两种适配CDK场景的常用实现方式：

方案1：辅助配置函数（推荐） #

这种方式无需修改SDK原生行为，逻辑透明易排查，适配绝大多数复用场景：

第一步：封装通用配置 #

你可以在公共工具文件中定义配置生成函数，支持单独返回公共访问阻断配置，或是把其他通用安全配置一起封装成可直接解包的参数字典：

from aws_cdk import aws_s3 as s3
import aws_cdk as cdk

# 仅封装公共访问阻断配置
def get_secure_block_public_access() -> s3.BlockPublicAccess:
    return s3.BlockPublicAccess(
        block_public_acls=True,
        block_public_policy=True,
        ignore_public_acls=True,
        restrict_public_buckets=True
    )

# 封装全量通用安全配置（可根据业务需求扩展）
def get_secure_bucket_base_config() -> dict:
    return {
        "block_public_access": get_secure_block_public_access(),
        "public_read_access": False,
        "encryption": s3.BucketEncryption.S3_MANAGED
    }

第二步：业务代码中复用 #

创建存储桶时直接引入使用即可，自定义配置可以覆盖通用配置：

def __create_s3_components_bucket(self, this_dir: str, props):
    """Create S3 components bucket"""
    s3_bucket = s3.Bucket(
        self,
        "BucketForImageBuilder",
        bucket_name="some_bucket_name_1234",
        # 解包通用安全配置
        **get_secure_bucket_base_config(),
        # 自定义业务配置，和通用配置重名时会自动覆盖
        removal_policy=cdk.RemovalPolicy.DESTROY,
        auto_delete_objects=True,
        lifecycle_rules=[
            s3.LifecycleRule(
                abort_incomplete_multipart_upload_after=cdk.Duration.days(amount=2),
                enabled=True,
                expiration=cdk.Duration.days(amount=180),
                transitions=[
                    s3.Transition(
                        transition_after=cdk.Duration.days(amount=30),
                        storage_class=s3.StorageClass.ONE_ZONE_INFREQUENT_ACCESS,
                    )
                ],
            )
        ],
    )

方案2：装饰器实现 #

如果需要强制所有栈内的存储桶都默认加上安全配置，避免遗漏，可以用装饰器拦截Bucket初始化逻辑自动注入参数：

from functools import wraps
from aws_cdk import aws_s3 as s3

def secure_s3_bucket_decorator(init_func):
    @wraps(init_func)
    def wrapper(*args, **kwargs):
        # 未传对应配置时自动注入安全默认值，已传的话保留用户自定义值
        if "block_public_access" not in kwargs:
            kwargs["block_public_access"] = s3.BlockPublicAccess(
                block_public_acls=True,
                block_public_policy=True,
                ignore_public_acls=True,
                restrict_public_buckets=True
            )
        if "public_read_access" not in kwargs:
            kwargs["public_read_access"] = False
        if "encryption" not in kwargs:
            kwargs["encryption"] = s3.BucketEncryption.S3_MANAGED
        return init_func(*args, **kwargs)
    return wrapper

# 全局替换s3.Bucket的初始化方法，后续所有创建桶的操作都会自动应用配置
s3.Bucket.__init__ = secure_s3_bucket_decorator(s3.Bucket.__init__)

装饰后创建存储桶时无需手动传通用安全参数即可自动生效，有特殊需求时手动传参即可覆盖默认值。

内容的提问来源于stack exchange，提问作者Maciej