Add new bucket policies; update docs

This commit is contained in:
2026-01-14 22:05:31 +08:00
parent 5522f9ac04
commit 956d17a649
5 changed files with 259 additions and 30 deletions

View File

@@ -72,13 +72,11 @@ def _evaluate_condition_operator(
return True
ACTION_ALIASES = {
# List actions
"s3:listbucket": "list",
"s3:listallmybuckets": "list",
"s3:listbucketversions": "list",
"s3:listmultipartuploads": "list",
"s3:listparts": "list",
# Read actions
"s3:getobject": "read",
"s3:getobjectversion": "read",
"s3:getobjecttagging": "read",
@@ -87,7 +85,6 @@ ACTION_ALIASES = {
"s3:getbucketversioning": "read",
"s3:headobject": "read",
"s3:headbucket": "read",
# Write actions
"s3:putobject": "write",
"s3:createbucket": "write",
"s3:putobjecttagging": "write",
@@ -97,26 +94,30 @@ ACTION_ALIASES = {
"s3:completemultipartupload": "write",
"s3:abortmultipartupload": "write",
"s3:copyobject": "write",
# Delete actions
"s3:deleteobject": "delete",
"s3:deleteobjectversion": "delete",
"s3:deletebucket": "delete",
"s3:deleteobjecttagging": "delete",
# Share actions (ACL)
"s3:putobjectacl": "share",
"s3:putbucketacl": "share",
"s3:getbucketacl": "share",
# Policy actions
"s3:putbucketpolicy": "policy",
"s3:getbucketpolicy": "policy",
"s3:deletebucketpolicy": "policy",
# Replication actions
"s3:getreplicationconfiguration": "replication",
"s3:putreplicationconfiguration": "replication",
"s3:deletereplicationconfiguration": "replication",
"s3:replicateobject": "replication",
"s3:replicatetags": "replication",
"s3:replicatedelete": "replication",
"s3:getlifecycleconfiguration": "lifecycle",
"s3:putlifecycleconfiguration": "lifecycle",
"s3:deletelifecycleconfiguration": "lifecycle",
"s3:getbucketlifecycle": "lifecycle",
"s3:putbucketlifecycle": "lifecycle",
"s3:getbucketcors": "cors",
"s3:putbucketcors": "cors",
"s3:deletebucketcors": "cors",
}

View File

@@ -15,7 +15,7 @@ class IamError(RuntimeError):
"""Raised when authentication or authorization fails."""
S3_ACTIONS = {"list", "read", "write", "delete", "share", "policy", "replication"}
S3_ACTIONS = {"list", "read", "write", "delete", "share", "policy", "replication", "lifecycle", "cors"}
IAM_ACTIONS = {
"iam:list_users",
"iam:create_user",
@@ -71,6 +71,16 @@ ACTION_ALIASES = {
"s3:replicateobject": "replication",
"s3:replicatetags": "replication",
"s3:replicatedelete": "replication",
"lifecycle": "lifecycle",
"s3:getlifecycleconfiguration": "lifecycle",
"s3:putlifecycleconfiguration": "lifecycle",
"s3:deletelifecycleconfiguration": "lifecycle",
"s3:getbucketlifecycle": "lifecycle",
"s3:putbucketlifecycle": "lifecycle",
"cors": "cors",
"s3:getbucketcors": "cors",
"s3:putbucketcors": "cors",
"s3:deletebucketcors": "cors",
"iam:listusers": "iam:list_users",
"iam:createuser": "iam:create_user",
"iam:deleteuser": "iam:delete_user",

View File

@@ -381,6 +381,23 @@ def bucket_detail(bucket_name: str):
can_edit_policy = True
except IamError:
can_edit_policy = False
can_manage_lifecycle = False
if principal:
try:
_iam().authorize(principal, bucket_name, "lifecycle")
can_manage_lifecycle = True
except IamError:
can_manage_lifecycle = False
can_manage_cors = False
if principal:
try:
_iam().authorize(principal, bucket_name, "cors")
can_manage_cors = True
except IamError:
can_manage_cors = False
try:
versioning_enabled = storage.is_versioning_enabled(bucket_name)
except StorageError:
@@ -452,6 +469,8 @@ def bucket_detail(bucket_name: str):
bucket_policy_text=policy_text,
bucket_policy=bucket_policy,
can_edit_policy=can_edit_policy,
can_manage_lifecycle=can_manage_lifecycle,
can_manage_cors=can_manage_cors,
can_manage_versioning=can_manage_versioning,
can_manage_replication=can_manage_replication,
can_manage_encryption=can_manage_encryption,
@@ -2128,7 +2147,7 @@ def metrics_api():
def bucket_lifecycle(bucket_name: str):
principal = _current_principal()
try:
_authorize_ui(principal, bucket_name, "policy")
_authorize_ui(principal, bucket_name, "lifecycle")
except IamError as exc:
return jsonify({"error": str(exc)}), 403
@@ -2181,7 +2200,7 @@ def bucket_lifecycle(bucket_name: str):
def get_lifecycle_history(bucket_name: str):
principal = _current_principal()
try:
_authorize_ui(principal, bucket_name, "policy")
_authorize_ui(principal, bucket_name, "lifecycle")
except IamError:
return jsonify({"error": "Access denied"}), 403
@@ -2212,7 +2231,7 @@ def get_lifecycle_history(bucket_name: str):
def bucket_cors(bucket_name: str):
principal = _current_principal()
try:
_authorize_ui(principal, bucket_name, "policy")
_authorize_ui(principal, bucket_name, "cors")
except IamError as exc:
return jsonify({"error": str(exc)}), 403