Skip to content

S3 Bucket Policy

S3BucketPolicyUtility

Bases: UtilitiesInterface

Assigns an IAM Bucket Policy to an S3 Bucket

Parameters:

Name Type Description Default
bucket_name str

S3 Bucket Name

 required
aws_access_key_id str

AWS Access Key

 required
aws_secret_access_key str

AWS Secret Key

 required
aws_session_token str

AWS Session Token

 required
sid str

S3 Bucket Policy Sid to be updated

 required
effect str

Effect to be applied to the policy

 required
principal str

Principal to be applied to Policy

 required
resource list[str]

List of resources to be applied to the policy

 required
Source code in src/sdk/python/rtdip_sdk/pipelines/utilities/aws/s3_bucket_policy.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class S3BucketPolicyUtility(UtilitiesInterface):
    """
    Assigns an IAM Bucket Policy to an S3 Bucket

    Args:
        bucket_name (str): S3 Bucket Name
        aws_access_key_id (str): AWS Access Key
        aws_secret_access_key (str): AWS Secret Key
        aws_session_token (str): AWS Session Token
        sid (str): S3 Bucket Policy Sid to be updated
        effect (str): Effect to be applied to the policy
        principal (str): Principal to be applied to Policy
        resource (list[str]): List of resources to be applied to the policy
    """

    bucket_name: str
    aws_access_key_id: str
    aws_secret_access_key: str
    aws_session_token: str
    sid: str
    effect: str
    principal: str
    action: List[str]
    resource: List[str]

    def __init__(
        self,
        bucket_name: str,
        aws_access_key_id: str,
        aws_secret_access_key: str,
        aws_session_token: str,
        sid: str,
        principal: str,
        effect: str,
        action: List[str],
        resource: List[str],
    ) -> None:
        self.bucket_name = bucket_name
        self.aws_access_key_id = aws_access_key_id
        self.aws_secret_access_key = aws_secret_access_key
        self.aws_session_token = aws_session_token
        self.sid = sid
        self.effect = effect
        self.principal = principal
        self.action = action
        self.resource = resource

    @staticmethod
    def system_type():
        """
        Attributes:
            SystemType (Environment): Requires PYTHON
        """
        return SystemType.PYTHON

    @staticmethod
    def libraries():
        libraries = Libraries()
        libraries.add_pypi_library(get_default_package("aws_boto3"))
        return libraries

    @staticmethod
    def settings() -> dict:
        return {}

    def execute(self) -> bool:
        try:
            s3_client = boto3.client(
                "s3",
                aws_access_key_id=self.aws_access_key_id,
                aws_secret_access_key=self.aws_secret_access_key,
                aws_session_token=self.aws_session_token,
            )

            bucket_policy = s3_client.get_bucket_policy(Bucket=self.bucket_name)

            policy_statement = None
            if "Policy" in bucket_policy and bucket_policy["Policy"] != None:
                policy_statement = json.loads(bucket_policy["Policy"])

            if policy_statement is None:
                policy_statement = {"Version": "2012-10-17", "Statement": []}

            sid_found = False
            for statement in policy_statement["Statement"]:
                if statement["Sid"] == self.sid:
                    sid_found = True
                    statement["Effect"] = self.effect
                    statement["Principal"] = self.principal
                    statement["Action"] = self.action
                    if isinstance(statement["Resource"], list):
                        statement["Resource"] + self.resource
                    else:
                        self.resource.append(statement["Resource"])
                        statement["Resource"] = self.resource
                    statement["Resource"] = list(set(statement["Resource"]))

            if not sid_found:
                policy_statement["Statement"].append(
                    {
                        "Sid": self.sid,
                        "Effect": self.effect,
                        "Principal": self.principal,
                        "Action": self.action,
                        "Resource": self.resource,
                    }
                )

            policy = json.dumps(policy_statement)
            s3_client.put_bucket_policy(Bucket=self.bucket_name, Policy=policy)

            return True

        except Exception as e:
            logging.exception(str(e))
            raise e

system_type() staticmethod

Attributes:

Name Type Description
SystemType Environment

Requires PYTHON
Source code in src/sdk/python/rtdip_sdk/pipelines/utilities/aws/s3_bucket_policy.py 
75
76
77
78
79
80
81
@staticmethod
def system_type():
    """
    Attributes:
        SystemType (Environment): Requires PYTHON
    """
    return SystemType.PYTHON