8000 Add storage acl samples · comcomhanb/python-docs-samples@9f9705a · GitHub
[go: up one dir, main page]
Skip to content

Commit 9f9705a

Browse files
author
Jon Wayne Parrott
committed
Add storage acl samples
Change-Id: Ib44f9bb42bf0c0607e64905a26369f06ea5fb231
1 parent f29f26d commit 9f9705a

File tree

2 files changed

+390
-0
lines changed

2 files changed

+390
-0
lines changed

‎storage/cloud-client/acl.py

Lines changed: 255 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,255 @@
1+
#!/usr/bin/env python
2+
3+
# Copyright 2016 Google, Inc.
4+
#
5+
# Licensed under the Apache License, Version 2.0 (the "License");
6+
# you may not use this file except in compliance with the License.
7+
# You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
17+
"""This application demonstrates how to manage access control lists (acls) in
18+
Google Cloud Storage.
19+
20+
For more information, see the README.md under /storage and the documentation
21+
at https://cloud.google.com/storage/docs/encryption.
22+
"""
23+
24+
import argparse
25+
26+
from gcloud import storage
27+
28+
29+
def get_bucket_acl(bucket_name):
30+
"""Prints out a bucket's access control list."""
31+
storage_client = storage.Client()
32+
bucket = storage_client.bucket(bucket_name)
33+
34+
for entry in bucket.acl:
35+
print('{}: {}'.format(entry['role'], entry['entity']))
36+
37+
38+
def get_bucket_acl_for_user(bucket_name, user_email):
39+
"""Prints out a bucket's access control list for a given user."""
40+
storage_client = storage.Client()
41+
bucket = storage_client.bucket(bucket_name)
42+
43+
bucket.acl.reload()
44+
45+
# You can also use `group`, `domain`, `all_authenticated` and `all` to
46+
# get the roles for different types of entities.
47+
roles = bucket.acl.user(user_email).get_roles()
48+
49+
print(roles)
50+
51+
52+
def set_bucket_acl(bucket_name, user_email):
53+
"""Adds a user as an owner on the given bucket."""
54+
storage_client = storage.Client()
55+
bucket = storage_client.bucket(bucket_name)
56+
57+
# You can also use `group`, `domain`, `all_authenticated` and `all` to
58+
# grant access to different types of entities. You can also use
59+
# `grant_read` or `grant_write` to grant different roles.
60+
bucket.acl.user(user_email).grant_owner()
61+
bucket.acl.save()
62+
63+
print('Added user {} as an owner on bucket {}.'.format(
64+
user_email, bucket_name))
65+
66+
67+
def remove_bucket_acl(bucket_name, user_email):
68+
"""Removes a user from the access control list of the given bucket."""
69+
storage_client = storage.Client()
70+
bucket = storage_client.bucket(bucket_name)
71+
72+
# You can also use `group`, `domain`, `all_authenticated` and `all` to
73+
# remove access for different types of entities.
74+
bucket.acl.user(user_email).revoke_read()
75+
bucket.acl.user(user_email).revoke_write()
76+
bucket.acl.user(user_email).revoke_owner()
77+
bucket.acl.save()
78+
79+
print('Removed user {} from bucket {}.'.format(
80+
user_email, bucket_name))
81+
82+
83+
def set_bucket_default_acl(bucket_name, user_email):
84+
"""Adds a user as an owner in the given bucket's default object access
85+
control list."""
86+
storage_client = storage.Client()
87+
bucket = storage_client.bucket(bucket_name)
88+
89+
# You can also use `group`, `domain`, `all_authenticated` and `all` to
90+
# grant access to different types of entities. You can also use
91+
# `grant_read` or `grant_write` to grant different roles.
92+
bucket.default_object_acl.user(user_email).grant_owner()
93+
bucket.default_object_acl.save()
94+
95+
print('Added user {} as an owner in the default acl on bucket {}.'.format(
96+
user_email, bucket_name))
97+
98+
99+
def remove_bucket_default_acl(bucket_name, user_email):
100+
"""Removes a user from the access control list of the given bucket's
101+
default object access control list."""
102+
storage_client = storage.Client()
103+
bucket = storage_client.bucket(bucket_name)
104+
105+
# You can also use `group`, `domain`, `all_authenticated` and `all` to
106+
# remove access for different types of entities.
107+
bucket.default_object_acl.user(user_email).revoke_read()
108+
bucket.default_object_acl.user(user_email).revoke_write()
109+
bucket.default_object_acl.user(user_email).revoke_owner()
110+
bucket.default_object_acl.save()
111+
112+
print('Removed user {} from the default acl of bucket {}.'.format(
113+
user_email, bucket_name))
114+
115+
116+
def get_blob_acl(bucket_name, blob_name):
117+
"""Prints out a blob's access control list."""
118+
storage_client = storage.Client()
119+
bucket = storage_client.bucket(bucket_name)
120+
blob = bucket.blob(blob_name)
121+
122+
for entry in blob.acl:
123+
print('{}: {}'.format(entry['role'], entry['entity']))
124+
125+
126+
def get_blob_acl_for_user(bucket_name, blob_name, user_email):
127+
"""Prints out a bucket's access control list for a given user."""
128+
storage_client = storage.Client()
129+
bucket = storage_client.bucket(bucket_name)
130+
blob = bucket.blob(blob_name)
131+
132+
blob.acl.reload()
133+
134+
# You can also use `group`, `domain`, `all_authenticated` and `all` to
135+
# get the roles for different types of entities.
136+
roles = blob.acl.user(user_email).get_roles()
137+
138+
print(roles)
139+
140+
141+
def set_blob_acl(bucket_name, blob_name, user_email):
142+
"""Adds a user as an owner on the given blob."""
143+
storage_client = storage.Client()
144+
bucket = storage_client.bucket(bucket_name)
145+
blob = bucket.blob(blob_name)
146+
147+
# You can also use `group`, `domain`, `all_authenticated` and `all` to
148+
# grant access to different types of entities. You can also use
149+
# `grant_read` or `grant_write` to grant different roles.
150+
blob.acl.user(user_email).grant_owner()
151+
blob.acl.save()
152+
153+
print('Added user {} as an owner on blob {} in bucket {}.'.format(
154+
user_email, blob_name, bucket_name))
155+
156+
157+
def remove_blob_acl(bucket_name, blob_name, user_email):
158+
"""Removes a user from the access control list of the given blob in the
159+
given bucket."""
160+
storage_client = storage.Client()
161+
bucket = storage_client.bucket(bucket_name)
162+
blob = bucket.blob(blob_name)
163+
164+
# You can also use `group`, `domain`, `all_authenticated` and `all` to
165+
# remove access for different types of entities.
166+
blob.acl.user(user_email).revoke_read()
167+
blob.acl.user(user_email).revoke_write()
168+
blob.acl.user(user_email).revoke_owner()
169+
blob.acl.save()
170+
171+
print('Removed user {} from blob {} in bucket {}.'.format(
172+
user_email, blob_name, bucket_name))
173+
174+
175+
if __name__ == '__main__':
176+
parser = argparse.ArgumentParser(
177+
description=__doc__,
178+
formatter_class=argparse.RawDescriptionHelpFormatter)
179+
subparsers = parser.add_subparsers(dest='command')
180+
181+
get_bucket_acl_parser = subparsers.add_parser(
182+
'get-bucket-acl', help=get_bucket_acl.__doc__)
183+
get_bucket_acl_parser.add_argument('bucket_name')
184+
185+
get_bucket_acl_for_user_parser = subparsers.add_parser(
186+
'get-bucket-acl-for-user', help=get_bucket_acl.__doc__)
187+
get_bucket_acl_for_user_parser.add_argument('bucket_name')
188+
get_bucket_acl_for_user_parser.add_argument('user_email')
189+
190+
set_bucket_acl_parser = subparsers.add_parser(
191+
'set-bucket-acl', help=set_bucket_acl.__doc__)
192+
set_bucket_acl_parser.add_argument('bucket_name')
193+
set_bucket_acl_parser.add_argument('user_email')
194+
195+
remove_bucket_acl_parser = subparsers.add_parser(
196+
'remove-bucket-acl', help=remove_bucket_acl.__doc__)
197+
remove_bucket_acl_parser.add_argument('bucket_name')
198+
remove_bucket_acl_parser.add_argument('user_email')
199+
200+
set_bucket_default_acl_parser = subparsers.add_parser(
201+
'set-bucket-default-acl', help=set_bucket_default_acl.__doc__)
202+
set_bucket_default_acl_parser.add_argument('bucket_name')
203+
set_bucket_default_acl_parser.add_argument('user_email')
204+
205+
remove_bucket_default_acl_parser = subparsers.add_parser(
206+
'remove-bucket-default-acl', help=remove_bucket_default_acl.__doc__)
207+
remove_bucket_default_acl_parser.add_argument('bucket_name')
208+
remove_bucket_default_acl_parser.add_argument('user_email')
209+
210+
get_blob_acl_parser = subparsers.add_parser(
211+
'get-blob-acl', help=get_blob_acl.__doc__)
212+
get_blob_acl_parser.add_argument('bucket_name')
213+
get_blob_acl_parser.add_argument('blob_name')
214+
215+
get_blob_acl_for_user_parser = subparsers.add_parser(
216+
'get-blob-acl-for-user', help=get_blob_acl_for_user.__doc__)
217+
get_blob_acl_for_user_parser.add_argument('bucket_name')
218+
get_blob_acl_for_user_parser.add_argument('blob_name')
219+
get_blob_acl_for_user_parser.add_argument('user_email')
220+
221+
set_blob_acl_parser = subparsers.add_parser(
222+
'set-blob-acl', help=set_blob_acl.__doc__)
223+
set_blob_acl_parser.add_argument('bucket_name')
224+
set_blob_acl_parser.add_argument('blob_name')
225+
set_blob_acl_parser.add_argument('user_email')
226+
227+
remove_blob_acl_parser = subparsers.add_parser(
228+
'remove-blob-acl', help=remove_blob_acl.__doc__)
229+
remove_blob_acl_parser.add_argument('bucket_name')
230+
remove_blob_acl_parser.add_argument('blob_name')
231+
remove_blob_acl_parser.add_argument('user_email')
232+
233+
args = parser.parse_args()
234+
235+
if args.command == 'get-bucket-acl':
236+
get_bucket_acl(args.bucket_name)
237+
elif args.command == 'get-bucket-acl-for-user':
238+
get_bucket_acl_for_user(args.bucket_name, args.user_email)
239+
elif args.command == 'set-bucket-acl':
240+
set_bucket_acl(args.bucket_name, args.user_email)
241+
elif args.command == 'remove-bucket-acl':
242+
remove_bucket_acl(args.bucket_name, args.user_email)
243+
elif args.command == 'set-bucket-default-acl':
244+
set_bucket_default_acl(args.bucket_name, args.user_email)
245+
elif args.command == 'remove-bucket-default-acl':
246+
remove_bucket_default_acl(args.bucket_name, args.user_email)
247+
elif args.command == 'get-blob-acl':
248+
get_blob_acl(args.bucket_name, args.blob_name)
249+
elif args.command == 'get-blob-acl-for-user':
250+
get_blob_acl_for_user(
251+
args.bucket_name, args.blob_name, args.user_email)
252+
elif args.command == 'set-blob-acl':
253+
set_blob_acl(args.bucket_name, args.blob_name, args.user_email)
254+
elif args.command == 'remove-blob-acl':
255+
remove_blob_acl(args.bucket_name, args.blob_name, args.user_email)

‎storage/cloud-client/acl_test.py

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
# Copyright 2016 Google, Inc.
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
from gcloud import storage
16+
import gcloud.storage.acl
17+
import pytest
18+
19+
import acl
20+
21+
# Typically we'd use a @example.com address, but GCS requires a real Google
22+
# account.
23+
TEST_EMAIL = 'jonwayne@google.com'
24+
25+
26+
@pytest.fixture
27+
def test_bucket(cloud_config):
28+
"""Yields a bucket that resets its acl after the test completes."""
29+
bucket = storage.Client().bucket(cloud_config.storage_bucket)
30+
acl = gcloud.storage.acl.BucketACL(bucket)
31+
object_default_acl = gcloud.storage.acl.DefaultObjectACL(bucket)
32+
acl.reload()
33+
object_default_acl.reload()
34+
yield bucket
35+
acl.save()
36+
object_default_acl.save()
37+
38+
39+
@pytest.fixture
40+
def test_blob(cloud_config):
41+
"""Yields a blob that resets its acl after the test completes."""
42+
bucket = storage.Client().bucket(cloud_config.storage_bucket)
43+
blob = bucket.blob('storage_acl_test_sigil')
44+
blob.upload_from_string('Hello, is it me you\'re looking for?')
45+
acl = gcloud.storage.acl.ObjectACL(blob)
46+
acl.reload()
47+
yield blob
48+
acl.save()
49+
50+
51+
def test_get_bucket_acl(cloud_config, capsys):
52+
acl.get_bucket_acl(cloud 10000 _config.storage_bucket)
53+
out, _ = capsys.readouterr()
54+
assert out
55+
56+
57+
def test_get_bucket_acl_for_user(test_bucket, cloud_config, capsys):
58+
test_bucket.acl.user(TEST_EMAIL).grant_owner()
59+
test_bucket.acl.save()
60+
61+
acl.get_bucket_acl_for_user(cloud_config.storage_bucket, TEST_EMAIL)
62+
63+
out, _ = capsys.readouterr()
64+
assert 'OWNER' in out
65+
66+
67+
def test_set_bucket_acl(test_bucket, cloud_config):
68+
acl.set_bucket_acl(cloud_config.storage_bucket, TEST_EMAIL)
69+
70+
test_bucket.acl.reload()
71+
assert 'OWNER' in test_bucket.acl.user(TEST_EMAIL).get_roles()
72+
73+
74+
def test_remove_bucket_acl(test_bucket, cloud_config):
75+
test_bucket.acl.user(TEST_EMAIL).grant_owner()
76+
test_bucket.acl.save()
77+
78+
acl.remove_bucket_acl(cloud_config.storage_bucket, TEST_EMAIL)
79+
80+
test_bucket.acl.reload()
81+
assert 'OWNER' not in test_bucket.acl.user(TEST_EMAIL).get_roles()
82+
83+
84+
def test_set_bucket_default_acl(test_bucket, cloud_config):
85+
acl.set_bucket_default_acl(cloud_config.storage_bucket, TEST_EMAIL)
86+
87+
test_bucket.default_object_acl.reload()
88+
roles = test_bucket.default_object_acl.user(TEST_EMAIL).get_roles()
89+
assert 'OWNER' in roles
90+
91+
92+
def test_remove_bucket_default_acl(test_bucket, cloud_config):
93+
test_bucket.acl.user(TEST_EMAIL).grant_owner()
94+
test_bucket.acl.save()
95+
96+
acl.remove_bucket_default_a 10000 cl(cloud_config.storage_bucket, TEST_EMAIL)
97+
98+
test_bucket.default_object_acl.reload()
99+
roles = test_bucket.default_object_acl.user(TEST_EMAIL).get_roles()
100+
assert 'OWNER' not in roles
101+
102+
103+
def test_get_blob_acl(test_blob, cloud_config, capsys):
104+
acl.get_blob_acl(cloud_config.storage_bucket, test_blob.name)
105+
out, _ = capsys.readouterr()
106+
assert out
107+
108+
109+
def test_get_blob_acl_for_user(test_blob, cloud_config, capsys):
110+
test_blob.acl.user(TEST_EMAIL).grant_owner()
111+
test_blob.acl.save()
112+
113+
acl.get_blob_acl_for_user(
114+
cloud_config.storage_bucket, test_blob.name, TEST_EMAIL)
115+
116+
out, _ = capsys.readouterr()
117+
assert 'OWNER' in out
118+
119+
120+
def test_set_blob_acl(test_blob, cloud_config):
121+
acl.set_blob_acl(cloud_config.storage_bucket, test_blob.name, TEST_EMAIL)
122+
123+
test_blob.acl.reload()
124+
assert 'OWNER' in test_blob.acl.user(TEST_EMAIL).get_roles()
125+
126+
127+
def test_remove_blob_acl(test_blob, cloud_config):
128+
test_blob.acl.user(TEST_EMAIL).grant_owner()
129+
test_blob.acl.save()
130+
131+
acl.remove_blob_acl(
132+
cloud_config.storage_bucket, test_blob.name, TEST_EMAIL)
133+
134+
test_blob.acl.reload()
135+
assert 'OWNER' not in test_blob.acl.user(TEST_EMAIL).get_roles()

0 commit comments

Comments
 (0)
0