-
-
Notifications
You must be signed in to change notification settings - Fork 11
/
policies.py
115 lines (100 loc) · 3.38 KB
/
policies.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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
def read_write(bucket, prefix="*", extra_statements=None):
statements = read_write_statements(bucket, prefix=prefix)
if extra_statements:
statements.extend(extra_statements)
return wrap_policy(statements)
def read_write_statements(bucket, prefix="*"):
# https://github.com/simonw/s3-credentials/issues/24
if not prefix.endswith("*"):
prefix += "*"
return read_only_statements(bucket, prefix) + [
{
"Effect": "Allow",
"Action": ["s3:PutObject", "s3:DeleteObject"],
"Resource": ["arn:aws:s3:::{}/{}".format(bucket, prefix)],
}
]
def read_only(bucket, prefix="*", extra_statements=None):
statements = read_only_statements(bucket, prefix=prefix)
if extra_statements:
statements.extend(extra_statements)
return wrap_policy(statements)
def read_only_statements(bucket, prefix="*"):
# https://github.com/simonw/s3-credentials/issues/23
statements = []
if not prefix.endswith("*"):
prefix += "*"
if prefix != "*":
statements.append(
{
"Effect": "Allow",
"Action": ["s3:GetBucketLocation"],
"Resource": ["arn:aws:s3:::{}".format(bucket)],
}
)
statements.append(
{
"Effect": "Allow",
"Action": ["s3:ListBucket"],
"Resource": ["arn:aws:s3:::{}".format(bucket)],
"Condition": {
"StringLike": {
# Note that prefix must end in / if user wants to limit to a folder
"s3:prefix": [prefix]
}
},
}
)
else:
# We can combine s3:GetBucketLocation and s3:ListBucket into one
statements.append(
{
"Effect": "Allow",
"Action": ["s3:ListBucket", "s3:GetBucketLocation"],
"Resource": ["arn:aws:s3:::{}".format(bucket)],
}
)
return statements + [
{
"Effect": "Allow",
"Action": [
"s3:GetObject",
"s3:GetObjectAcl",
"s3:GetObjectLegalHold",
"s3:GetObjectRetention",
"s3:GetObjectTagging",
],
"Resource": ["arn:aws:s3:::{}/{}".format(bucket, prefix)],
},
]
def write_only(bucket, prefix="*", extra_statements=None):
statements = write_only_statements(bucket, prefix=prefix)
if extra_statements:
statements.extend(extra_statements)
return wrap_policy(statements)
def write_only_statements(bucket, prefix="*"):
# https://github.com/simonw/s3-credentials/issues/25
if not prefix.endswith("*"):
prefix += "*"
return [
{
"Effect": "Allow",
"Action": ["s3:PutObject"],
"Resource": ["arn:aws:s3:::{}/{}".format(bucket, prefix)],
}
]
def wrap_policy(statements):
return {"Version": "2012-10-17", "Statement": statements}
def bucket_policy_allow_all_get(bucket):
return {
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowAllGetObject",
"Effect": "Allow",
"Principal": "*",
"Action": ["s3:GetObject"],
"Resource": ["arn:aws:s3:::{}/*".format(bucket)],
}
],
}