Set expiration for policy statements

Condition key

Date operators

acs:CurrentTime supports the following six date operators:

Cleanup procedure

1. Call GetBucketPolicy to retrieve the complete Policy JSON text.

2. Parse the policy JSON and iterate through each statement.

3. Find DateLessThan or DateGreaterThan in Condition, and extract the time string for acs:CurrentTime.

4. Parse the time string into a datetime object and compare it with the current system time.

5. If a time condition will never be triggered again (for example, the specified time for DateLessThan is earlier than the current time), the Statement is marked as expired and removed from the list.

6. Call PutBucketPolicy to resubmit the list of Statements after the expired entries are removed.

Identifying expired statements

To determine whether an expired statement can be safely deleted, consider both the Condition operator and Effect:

Example Python automation script

This sample script uses the oss2 SDK to identify and remove expired statements with the DateLessThan condition. Adjust the rules for your business logic.

import json
import alibabacloud_oss_v2 as oss
from datetime import datetime, timezone
import dateutil.parser  # We recommend installing the python-dateutil library to correctly parse ISO 8601 strings.

# Configurations
access_key = 'YOUR_ACCESS_KEY'
secret_key = 'YOUR_SECRET_KEY'
region = 'cn-hangzhou'
bucket_name = 'your-bucket-name'

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. Get the current 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)  # Use UTC for time comparison.

        new_statements = []
        removed_count = 0

        # 2. Iterate through and check each statement.
        for stmt in statements:
            is_expired = False

            # Check for time-based conditions (this example uses DateLessThan).
            condition = stmt.get('Condition', {})
            date_less_than = condition.get('DateLessThan', {}).get('acs:CurrentTime')

            if date_less_than:
                # Extract the time from the policy (for example, "2026-08-12T17:00:00+08:00").
                # dateutil.parser can automatically handle time zone information, such as +08:00.
                expiry_time = dateutil.parser.isoparse(date_less_than[0])

                # If the specified time is earlier than the current time, the Deny/Allow statement will never be triggered again.
                if expiry_time < now:
                    is_expired = True
                    print(f"Found expired statement [Sid: {stmt.get('Sid', 'N/A')}]: The set time {date_less_than[0]} has passed.")

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

        # 3. Write the policy back to the bucket.
        if removed_count > 0:
            if not new_statements:
                # If all statements are expired, delete the entire policy.
                client.delete_bucket_policy(oss.models.DeleteBucketPolicyRequest(bucket=bucket_name))
                print("All statements have expired. The bucket policy has been cleared.")
            else:
                policy_data['Statement'] = new_statements
                client.put_bucket_policy(oss.models.PutBucketPolicyRequest(
                    bucket=bucket_name,
                    body=json.dumps(policy_data),
                ))
                print(f"Cleanup complete. Removed {removed_count} expired statement(s).")
        else:
            print("No expired statements found.")

    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("This bucket does not have a policy configured.")
        else:
            print(f"An error occurred: {e}")
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    clean_expired_policy()