aws_test_bucket.py 文件源码

def check_auth_bucket(bucket_name):
# Let's check S3 static web site hosting status
    try:
        website_status =s3.get_bucket_website(Bucket=bucket_name)
        bucket_status_code = "S3WebSite!"
        return bucket_status_code
    except ClientError as ex:
        bucket_status_code = ex.response['Error']['Code']
# Let's try to get bucket ACL and Policy
# ACL
    try:
        bucket_acl = s3.get_bucket_acl(Bucket=bucket_name)
        for grant in bucket_acl["Grants"]:
            if grant["Grantee"]["Type"] == "Group" and "AllUsers" in grant["Grantee"].get("URI"):
                bucket_status_code = "AllUsersAccess"
                return bucket_status_code
            elif grant["Grantee"]["Type"] == "Group" and "AuthenticatedUsers" in grant["Grantee"].get("URI"):
                bucket_status_code = "AllAuthUsersAccess"
                return bucket_status_code

    except ClientError as ex:
        if ex.response['Error']['Code'] == "AccessDenied":
            bucket_status_code = "AccessDenied2ACL"
        else:
            bucket_status_code ="Can'tVerify"
            # cprint ("Weird"+ str(ex.response['Error']), "red")
#Policy
    try:
        bucket_policy = s3.get_bucket_policy(Bucket=bucket_name)
        bucket_policy_j = json.loads(bucket_policy["Policy"])
        for statement in bucket_policy_j["Statement"]:
            if (statement.get("Condition") is None and
                statement["Effect"] == "Allow" and
                ("'*'" in str(statement["Principal"]) or statement["Principal"] == "*")):
                bucket_status_code = str(statement["Action"])
                return bucket_status_code
# Policy exists but not allow public access
        bucket_status_code = "NoPublicAccess"
    except ClientError as ex:
        if ex.response['Error']['Code'] == "NoSuchBucketPolicy":
            bucket_status_code = "NoSuchBucketPolicy"
        elif ex.response['Error']['Code'] == "AccessDenied":
            bucket_status_code = "AccessDenied2Policy"

        else:
            bucket_status_code ="Can'tVerify"
            # cprint("Weird"+ str(ex.response['Error']), "red")
#   return status code
    return bucket_status_code