为 Bucket Policy 和 RAM Policy 中的 Statement 设置有效期

条件键

日期类操作符

acs:CurrentTime 属于日期类型，支持以下 6 个操作符：

清理流程

1. 调用 GetBucketPolicy 获取完整的 Policy JSON 文本。

2. 解析 JSON，遍历每一条 Statement。

3. 查找 Condition 中的 DateLessThan 或 DateGreaterThan，提取 acs:CurrentTime 对应的时间字符串。

4. 将时间字符串解析为 datetime 对象，与当前系统时间对比。

5. 如果时间条件已经永远不会再触发（例如 DateLessThan 的指定时间早于当前时间），则将该 Statement 标记为过期并从列表中移除。

6. 调用 PutBucketPolicy 将剔除过期条目后的 Statement 列表重新提交。

如何识别"过期"

清理脚本判定过期时，需要结合 Condition 操作符和 Effect 综合理解业务意图：

Python 自动化脚本示例

以下示例脚本基于 oss2 SDK，演示如何识别并清理仅含 DateLessThan 时间条件的过期 Statement。请根据实际业务逻辑调整识别和清理规则。

import json
import alibabacloud_oss_v2 as oss
from datetime import datetime, timezone
import dateutil.parser  # 建议安装 python-dateutil 库来完美解析 ISO8601 格式

# 配置信息
access_key = '您的AccessKey'
secret_key = '您的SecretKey'
region = 'cn-hangzhou'
bucket_name = '您的Bucket名称'

cfg = oss.config.load_default()
cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(access_key, secret_key)
cfg.region = region
client = oss.Client(cfg)

def clean_expired_policy():
    try:
        # 1. 获取当前 Bucket Policy
        result = client.get_bucket_policy(oss.models.GetBucketPolicyRequest(bucket=bucket_name))
        policy_data = json.loads(result.body)

        statements = policy_data.get('Statement', [])
        now = datetime.now(timezone.utc)  # 使用 UTC 时间进行对比

        new_statements = []
        removed_count = 0

        # 2. 遍历并检查
        for stmt in statements:
            is_expired = False

            # 检查是否有时间限制条件 (以 DateLessThan 为例)
            condition = stmt.get('Condition', {})
            date_less_than = condition.get('DateLessThan', {}).get('acs:CurrentTime')

            if date_less_than:
                # 提取策略里的时间 (例如 "2026-08-12T17:00:00+08:00")
                # dateutil.parser 可以自动处理 +08:00 等时区信息
                expiry_time = dateutil.parser.isoparse(date_less_than[0])

                # 如果设定时间早于当前时间，说明该 Deny/Allow 已经再也不会触发了
                if expiry_time < now:
                    is_expired = True
                    print(f"发现失效 Statement [Sid: {stmt.get('Sid', 'N/A')}]: 设定时间 {date_less_than[0]} 已过期。")

            if not is_expired:
                new_statements.append(stmt)
            else:
                removed_count += 1

        # 3. 写回 Policy
        if removed_count > 0:
            if not new_statements:
                # 如果删空了，直接删除整个 Policy
                client.delete_bucket_policy(oss.models.DeleteBucketPolicyRequest(bucket=bucket_name))
                print("所有 Statement 已过期，已清空 Bucket Policy。")
            else:
                policy_data['Statement'] = new_statements
                client.put_bucket_policy(oss.models.PutBucketPolicyRequest(
                    bucket=bucket_name,
                    body=json.dumps(policy_data),
                ))
                print(f"清理完成，共移除 {removed_count} 条过期条目。")
        else:
            print("未发现过期条目。")

    except oss.exceptions.OperationError as e:
        cause = e.unwrap() if hasattr(e, 'unwrap') else None
        if isinstance(cause, oss.exceptions.ServiceError) and cause.code == 'NoSuchBucketPolicy':
            print("该 Bucket 没有配置 Policy。")
        else:
            print(f"发生错误: {e}")
    except Exception as e:
        print(f"发生错误: {e}")

if __name__ == "__main__":
    clean_expired_policy()