i have enabled replication for an s3 bucket using boto3 ,needed to enable KMS as well but when i try to test this functionality by uploading files manually in the source bucket, they don't get replicated in the destination bucket.
When i check the status of the replication on the console i can see that it's failing but we don't have the details on the cause.
FYI , we have an existing policy attached to the role i am using here and i updated it to reflect the bellow changes :
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": [
"greengrass:*",
"wafv2:ListWebACLs",
"iotevents:DeleteAlarmModel",
"cloudfront:ListCloudFrontOriginAccessIdentities",
"iotevents:UpdateAlarmModel",
"iot:*",
"cloudfront:DeleteCloudFrontOriginAccessIdentity",
"iotevents:ListAlarmModels",
"iotsitewise:*",
"iotevents:ListAlarms",
"cloudfront:Get*",
"iam:CreateRole",
"iotevents:DescribeAlarm",
"kinesisanalytics:DescribeApplication",
"ses:SendEmail",
"iam:PassRole",
"iam:ListAttachedRolePolicies",
"kms:DescribeKey",
"kms:ListKeys",
"kinesisanalytics:ListApplications",
"iotevents:DescribeAlarmModel",
"iam:GetRole",
"kms:GetKeyPolicy",
"iam:ListRoles",
"iotevents:CreateAlarmModel",
"firehose:DescribeDeliveryStream",
"kms:ListAliases",
"servicequotas:*",
"lambda:*",
"route53:*",
"kms:GenerateDataKey",
"cloudfront:ListDistributions",
"firehose:ListDeliveryStreams",
"iam:ListUsers",
"cloudfront:DeleteDistribution",
"kinesisanalytics:StartApplication",
"acm:*"
],
"Resource": "*"
},
{
"Sid": "VisualEditor1",
"Effect": "Allow",
"Action": [
"s3:GetAccountPublicAccessBlock",
"s3:ListAllMyBuckets",
"s3:PutBucketEncryption"
],
"Resource": "arn:aws:s3:::*"
},
{
"Sid": "VisualEditor2",
"Effect": "Allow",
"Action": [
"s3:GetObjectVersionForReplication",
"s3:GetObjectVersionAcl"
],
"Resource": [
"arn:aws:s3:::source-bucket/*"
]
},
{
"Sid": "VisualEditor3",
"Effect": "Allow",
"Action": [
"s3:ListBucket",
"s3:GetReplicationConfiguration"
],
"Resource": [
"arn:aws:s3:::source-bucket"
]
},
{
"Sid": "VisualEditor4",
"Effect": "Allow",
"Action": [
"s3:ReplicateObject",
"s3:ReplicateDelete",
"s3:ReplicateTags",
"s3:GetObjectVersionTagging"
],
"Resource": "arn:aws:s3:::destination-bucket"
},
{
"Sid": "VisualEditor5",
"Effect": "Allow",
"Action": [
"iam:GetRole",
"iam:PassRole"
],
"Resource": "arn:aws:iam::id:role/fakereplicationrole"
},
{
"Sid": "VisualEditor6",
"Effect": "Allow",
"Action": "sts:AssumeRole",
"Resource": "arn:aws:iam::id:role/fakereplicationrole"
}
]
}
Below is the code :
import boto3
import botocore
import logging
s3_client=boto3.client('s3')
kms_client = boto3.client('kms')
my_region=s3_client.meta.region_name
region = my_region
storageClass = 'STANDARD'
prefix_filter = ''
deleteMarkerReplication = 'Enabled'
existingObjectReplication = 'Enabled'
def get_role():
"""
Description : get the role needed to enable the replication
"""
response = boto3.client("iam").get_role(RoleName='fakerolename')
arn=response['Role']['Arn']
return arn
def get_versioning(bucket: str):
"""
Description :check if the versioning is enabled for the source bucket , if not the function enables it
Args : bucket(str)
"""
# GET VERSIONING OF SOURCE BUCKET
s3_client.get_bucket_versioning(Bucket=bucket)
logging.info(f'Bucket "{bucket}" : enabling versioning')
s3_client.put_bucket_versioning(Bucket=bucket, VersioningConfiguration={'Status': 'Enabled' })
def create_target_bucket(bucket: str):
"""
Description : create the replicated bucket
Args : bucket(str)
"""
logging.info('Creating Target Bucket')
s3_client.create_bucket( Bucket=f'{bucket}-target', CreateBucketConfiguration={'LocationConstraint': region,},)
logging.info(f'Creating versioning for the target bucket "{bucket}-target" ')
s3_client.put_bucket_versioning(Bucket=f'{bucket}-target', VersioningConfiguration={'Status': 'Enabled'},)
def enable_replication(bucket: str):
"""
Description : enable replication for the source bucket
Args : bucket(str)
"""
logging.info(f'Inserting replication for the bucket "{bucket}" ')
s3_client.put_bucket_replication(Bucket=bucket,
#Modify the entry below with your account and the role you created
ReplicationConfiguration={
"Role": get_role(),
"Rules": [
{
"Status": "Enabled",
"Priority": 1,
"Filter": {"Prefix": prefix_filter},
"DeleteMarkerReplication": {"Status": deleteMarkerReplication},
"Destination": {
"Bucket": "arn:aws:s3:::%s-target" % bucket,"StorageClass": storageClass
}
}
]
}
)
print(bucket)
def get_s3_kms_key():
"""
Description : get the aws kms key for s3
"""
response = kms_client.list_aliases(Limit=123)
target_key_id=[]
for element in response['Aliases']:
if 's3' in element['AliasName']:
target_key_id.append(element['TargetKeyId'])
return target_key_id[0]
def set_bucket_kms_encryption(bucket: str):
"""
Description : enable kms encryption for the source bucket
Args : bucket(str)
"""
s3_client.put_bucket_encryption(Bucket=bucket,
ServerSideEncryptionConfiguration={
'Rules': [
{
'ApplyServerSideEncryptionByDefault': {
'SSEAlgorithm': 'aws:kms',
'KMSMasterKeyID': get_s3_kms_key()
},
'BucketKeyEnabled': True
}
]
})
def create_s3_backupdata(bucketname: str):
"""
Description : create the source bucket , enable replication and kms
Args : bucketname(str)
"""
s3_client.create_bucket( Bucket=bucketname, CreateBucketConfiguration={'LocationConstraint': region})
set_bucket_kms_encryption(bucket=bucketname)
try:
logging.info(f'Bucket "{bucketname}" already had replication')
config=s3_client.get_bucket_replication(Bucket=bucketname)
logging.info(f'bucket config "{config}" ')
except botocore.exceptions.ClientError:
# Log replication config
logging.info(f'Bucket "{bucketname}" : Replication config not enabled. Will enable versioning on source bucket and create a replication config')
# GET VERSIONING
get_versioning(bucketname)
# CREATE TARGET BUCKET
create_target_bucket(bucketname)
# Checking if config was created and skipping if not needed
enable_replication(bucketname)
def main():
bucketname='test1-kms-replication'
create_s3_backupdata(bucketname)
if __name__ == '__main__':
main()
What am i missing ?