Skip to content

updating s3-create for contraint issue #150

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mbaldessari merged 1 commit into from
Jan 14, 2025
Merged

updating s3-create for contraint issue #150

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
232 changes: 102 additions & 130 deletions aws-tools/s3-create.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,170 +8,142 @@
import os
import json

regions=['us-west-1', 'us-west-2', 'us-east-1', 'us-east-2']
regions = ['us-west-1', 'us-west-2', 'us-east-1', 'us-east-2']

try:
aws_access_key_id = os.environ['AWS_ACCESS_KEY_ID']
aws_secret_access_key = os.environ['AWS_SECRET_ACCESS_KEY']
except:
print("Please make sure that you set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables for AWS access")
exit()

if aws_access_key_id == "" or aws_secret_access_key == "":
print("Please make sure that you set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables for AWS access")
exit()

# Remove 1st argument from the
# list of command line arguments
argumentList = sys.argv[1:]
aws_access_key_id = os.environ['AWS_ACCESS_KEY_ID']
aws_secret_access_key = os.environ['AWS_SECRET_ACCESS_KEY']
except KeyError:
print("Please set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables for AWS access.")
exit()

# Options
options = "hpb:r:"
if not aws_access_key_id or not aws_secret_access_key:
print("AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables cannot be empty.")
exit()

# Long options
argumentList = sys.argv[1:]
options = "hpb:r:"
long_options = ["help", "bucket", "region", "public"]

def usage():
msg = "Usage: python " + sys.argv[0] + "\n" \
"Options: \n" \
" -b or --bucket \n" \
" S3 bucket name to create \n" \
" Example: -b my-bucket \n" \
" -r or --region \n" \
" AWS region for bucket \n" \
" Example: -r us-west-1 \n" \
" -h or --help \n" \
" Usage message"
print(msg)
exit()

def getSTSClient ():
try:
sts = boto3.client('sts')
return sts
except ClientError:
print ("Could not create STS client")
raise

def getSTSSessionToken( sts ):
msg = ("Usage: python " + sys.argv[0] + "\n"
"Options: \n"
" -b or --bucket \n"
" S3 bucket name to create \n"
" Example: -b my-bucket \n"
" -r or --region \n"
" AWS region for bucket \n"
" Example: -r us-west-1 \n"
" -h or --help \n"
" Usage message")
print(msg)
exit()

def get_s3_client(region):
endpoint_url = f"https://s3.{region}.amazonaws.com"
try:
token = sts.get_session_token()
sessionToken = token['Credentials']['SessionToken']
return sessionToken
except ClientError:
print("Could not retrieve STS Session Token")
raise

def getS3Client(region):
endpoint_url = 'https://s3.' + region + '.amazonaws.com'

try:
s3 = boto3.client('s3',
endpoint_url = endpoint_url,
aws_access_key_id = aws_access_key_id,
aws_secret_access_key = aws_secret_access_key,
region_name = region,
config=botocore.client.Config(signature_version = 's3v4'))
s3 = boto3.client(
's3',
endpoint_url=endpoint_url,
aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key,
region_name=region,
config=botocore.client.Config(signature_version='s3v4')
)
return s3
except ClientError:
print("Could not retrieve S3 client")
print("Could not retrieve S3 client.")
raise

def getSNSClient(region):
endpoint_url = 'https://s3.' + region + '.amazonaws.com'

sts = getSTSClient()
sessionToken = getSTSSessionToken(sts)
def create_bucket(s3_client, bucket_name, region, public=None):
try:
sns = boto3.client('sns',
endpoint_url = endpoint_url,
aws_access_key_id = aws_access_key_id,
aws_secret_access_key= aws_secret_access_key,
aws_session_token=sessionToken,
region_name=region,
config=botocore.client.Config(signature_version = 's3'))
return sns
except ClientError:
print ("Could not retrieve SNS client")
raise

def create_bucket(s3, bucket_name, region, public=None):
try:
result = getS3Client(region).create_bucket(Bucket=bucket_name, CreateBucketConfiguration={'LocationConstraint': region})
# For us-east-1, do not include CreateBucketConfiguration
if region == 'us-east-1':
result = s3_client.create_bucket(Bucket=bucket_name)
else:
result = s3_client.create_bucket(
Bucket=bucket_name,
CreateBucketConfiguration={'LocationConstraint': region}
)

if public:
getS3Client(region).put_public_access_block(Bucket=bucket_name, PublicAccessBlockConfiguration={'BlockPublicAcls': False,'IgnorePublicAcls': False,'BlockPublicPolicy': False,'RestrictPublicBuckets': False})
arn=[]
arn.append("arn:aws:s3:::" + bucket_name)
arn.append("arn:aws:s3:::" + bucket_name + "/*")
policy={
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowALLStatement1",
"Effect": "Allow",
"Principal": "*",
"Action": [
"s3:*",
],
"Resource": arn
}
s3_client.put_public_access_block(
Bucket=bucket_name,
PublicAccessBlockConfiguration={
'BlockPublicAcls': False,
'IgnorePublicAcls': False,
'BlockPublicPolicy': False,
'RestrictPublicBuckets': False
}
)
arn = [
f"arn:aws:s3:::{bucket_name}",
f"arn:aws:s3:::{bucket_name}/*"
]
}
policy = {
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowALLStatement1",
"Effect": "Allow",
"Principal": "*",
"Action": [
"s3:*",
],
"Resource": arn
}
]
}
s3_client.put_bucket_policy(Bucket=bucket_name, Policy=json.dumps(policy))

return result

except ClientError as e:
print("Could not create bucket [" + bucket_name + "]")
print(f"Could not create bucket [{bucket_name}]")
print(e)
raise

def main():
bucket=""
region=""
public=None
bucket = ""
region = ""
public = None

try:
# Parsing argument
arguments, values = getopt.getopt(argumentList, options, long_options)
arguments, _ = getopt.getopt(argumentList, options, long_options)

if len(arguments) == 0:
if not arguments:
usage()

# checking each argument
for currentArgument, currentValue in arguments:

for currentArgument, currentValue in arguments:
if currentArgument in ("-h", "--help"):
usage()
break
elif currentArgument in ("-b", "--bucket"):
print ("Creating S3 Bucket: ", currentValue)
bucket=currentValue
print("Creating S3 Bucket: ", currentValue)
bucket = currentValue
elif currentArgument in ("-r", "--region"):
print ("Target S3 Bucket Region: ", currentValue)
region=currentValue
print("Target S3 Bucket Region: ", currentValue)
region = currentValue
elif currentArgument in ("-p", "--public"):
print ("WARNING: S3 Bucket will be made public.")
public=True
print("WARNING: S3 Bucket will be made public.")
public = True

except getopt.error as err:
# output error, and return with an error code
print (str(err))
print(str(err))
usage()

if not region:
print("Missing required region.")
usage()
exit()

if region == "":
print ("Missing required region")
usage()

if bucket == "":
print ("Missing required bucket name")
usage()

s3Client = getS3Client(region)
s3 = boto3.resource('s3')
sns = getSNSClient(region)
result = create_bucket(s3, bucket, region, public)

if not bucket:
print("Missing required bucket name.")
usage()

s3_client = get_s3_client(region)
result = create_bucket(s3_client, bucket, region, public)

if result:
print ("Bucket [" + bucket + "] created successfully")
print(f"Bucket [{bucket}] created successfully.")


if __name__ == "__main__":
main()
Loading