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
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re
import uuid
import google.auth
from google.auth import downscoped
from google.auth.transport import requests
from google.cloud import exceptions
from google.cloud import storage
from google.oauth2 import credentials
import pytest
# The object prefix used to test access to files beginning with this prefix.
_OBJECT_PREFIX = "customer-a"
# The object name of the object inaccessible by the downscoped token.
_ACCESSIBLE_OBJECT_NAME = "{0}-data.txt".format(_OBJECT_PREFIX)
# The content of the object accessible by the downscoped token.
_ACCESSIBLE_CONTENT = "hello world"
# The content of the object inaccessible by the downscoped token.
_INACCESSIBLE_CONTENT = "secret content"
# The object name of the object inaccessible by the downscoped token.
_INACCESSIBLE_OBJECT_NAME = "other-customer-data.txt"
@pytest.fixture(scope="module")
def temp_bucket():
"""Yields a bucket that is deleted after the test completes."""
bucket = None
while bucket is None or bucket.exists():
bucket_name = "auth-python-downscope-test-{}".format(uuid.uuid4())
bucket = storage.Client().bucket(bucket_name)
bucket = storage.Client().create_bucket(bucket.name)
yield bucket
bucket.delete(force=True)
@pytest.fixture(scope="module")
def temp_blobs(temp_bucket):
"""Yields two blobs that are deleted after the test completes."""
bucket = temp_bucket
# Downscoped tokens will have readonly access to this blob.
accessible_blob = bucket.blob(_ACCESSIBLE_OBJECT_NAME)
accessible_blob.upload_from_string(_ACCESSIBLE_CONTENT)
# Downscoped tokens will have no access to this blob.
inaccessible_blob = bucket.blob(_INACCESSIBLE_OBJECT_NAME)
inaccessible_blob.upload_from_string(_INACCESSIBLE_CONTENT)
yield (accessible_blob, inaccessible_blob)
bucket.delete_blobs([accessible_blob, inaccessible_blob])
def get_token_from_broker(bucket_name, object_prefix):
"""Simulates token broker generating downscoped tokens for specified bucket.
Args:
bucket_name (str): The name of the Cloud Storage bucket.
object_prefix (str): The prefix string of the object name. This is used
to ensure access is restricted to only objects starting with this
prefix string.
Returns:
Tuple[str, datetime.datetime]: The downscoped access token and its expiry date.
"""
# Initialize the Credential Access Boundary rules.
available_resource = "//storage.googleapis.com/projects/_/buckets/{0}".format(bucket_name)
# Downscoped credentials will have readonly access to the resource.
available_permissions = ["inRole:roles/storage.objectViewer"]
# Only objects starting with the specified prefix string in the object name
# will be allowed read access.
availability_expression = (
"resource.name.startsWith('projects/_/buckets/{0}/objects/{1}')".format(bucket_name, object_prefix)
)
availability_condition = downscoped.AvailabilityCondition(availability_expression)
# Define the single access boundary rule using the above properties.
rule = downscoped.AccessBoundaryRule(
available_resource=available_resource,
available_permissions=available_permissions,
availability_condition=availability_condition,
)
# Define the Credential Access Boundary with all the relevant rules.
credential_access_boundary = downscoped.CredentialAccessBoundary(rules=[rule])
# Retrieve the source credentials via ADC.
source_credentials, _ = google.auth.default()
if source_credentials.requires_scopes:
source_credentials = source_credentials.with_scopes(
["https://www.googleapis.com/auth/cloud-platform"]
)
# Create the downscoped credentials.
downscoped_credentials = downscoped.Credentials(
source_credentials=source_credentials,
credential_access_boundary=credential_access_boundary,
)
# Refresh the tokens.
downscoped_credentials.refresh(requests.Request())
# These values will need to be passed to the token consumer.
access_token = downscoped_credentials.token
expiry = downscoped_credentials.expiry
return (access_token, expiry)
def test_downscoping(temp_blobs):
"""Tests token consumer access to cloud storage using downscoped tokens.
Args:
temp_blobs (Tuple[google.cloud.storage.blob.Blob, ...]): The temporarily
created test cloud storage blobs (one readonly accessible, the other
not).
"""
accessible_blob, inaccessible_blob = temp_blobs
bucket_name = accessible_blob.bucket.name
# Create the OAuth credentials from the downscoped token and pass a
# refresh handler to handle token expiration. We are passing a
# refresh_handler instead of a one-time access token/expiry pair.
# This will allow testing this on-demand method for getting access tokens.
def refresh_handler(request, scopes=None):
# Get readonly access tokens to objects with accessible prefix in
# the temporarily created bucket.
return get_token_from_broker(bucket_name, _OBJECT_PREFIX)
creds = credentials.Credentials(
None,
scopes=["https://www.googleapis.com/auth/cloud-platform"],
refresh_handler=refresh_handler,
)
# Initialize a Cloud Storage client with the oauth2 credentials.
storage_client = storage.Client(credentials=creds)
# Test read access succeeds to accessible blob.
bucket = storage_client.bucket(bucket_name)
blob = bucket.blob(accessible_blob.name)
assert blob.download_as_bytes().decode("utf-8") == _ACCESSIBLE_CONTENT
# Test write access fails.
with pytest.raises(exceptions.Forbidden) as excinfo:
blob.upload_from_string("Write operations are not allowed")
assert excinfo.match(r"does not have storage.objects.create access")
# Test read access fails to inaccessible blob.
with pytest.raises(exceptions.Forbidden) as excinfo:
bucket.blob(inaccessible_blob.name).download_as_bytes()
assert excinfo.match(r"does not have storage.objects.get access")
|
