##// END OF EJS Templates
tests: added some more artifact access tests.
marcink -
r4008:786403b1 default
parent child Browse files
Show More
@@ -21,7 +21,8 b' import os'
21 21 import pytest
22 22
23 23 from rhodecode.lib.ext_json import json
24 from rhodecode.model.db import Session, FileStore
24 from rhodecode.model.auth_token import AuthTokenModel
25 from rhodecode.model.db import Session, FileStore, Repository, User
25 26 from rhodecode.tests import TestController
26 27 from rhodecode.apps.file_store import utils, config_keys
27 28
@@ -32,6 +33,7 b' def route_path(name, params=None, **kwar'
32 33 base_url = {
33 34 'upload_file': '/_file_store/upload',
34 35 'download_file': '/_file_store/download/{fid}',
36 'download_file_by_token': '/_file_store/token-download/{_auth_token}/{fid}'
35 37
36 38 }[name].format(**kwargs)
37 39
@@ -124,3 +126,136 b' class TestFileStoreViews(TestController)'
124 126 status=200)
125 127
126 128 assert response.json['store_fid']
129
130 @pytest.fixture()
131 def create_artifact_factory(self, tmpdir):
132 def factory(user_id, content):
133 store_path = self.app._pyramid_settings[config_keys.store_path]
134 store = utils.get_file_storage({config_keys.store_path: store_path})
135 fid = 'example.txt'
136
137 filesystem_file = os.path.join(str(tmpdir), fid)
138 with open(filesystem_file, 'wb') as f:
139 f.write(content)
140
141 with open(filesystem_file, 'rb') as f:
142 store_uid, metadata = store.save_file(f, fid, extra_metadata={'filename': fid})
143
144 entry = FileStore.create(
145 file_uid=store_uid, filename=metadata["filename"],
146 file_hash=metadata["sha256"], file_size=metadata["size"],
147 file_display_name='file_display_name',
148 file_description='repo artifact `{}`'.format(metadata["filename"]),
149 check_acl=True, user_id=user_id,
150 )
151 Session().add(entry)
152 Session().commit()
153 return entry
154 return factory
155
156 def test_download_file_non_scoped(self, user_util, create_artifact_factory):
157 user = self.log_user()
158 user_id = user['user_id']
159 content = 'HELLO MY NAME IS ARTIFACT !'
160
161 artifact = create_artifact_factory(user_id, content)
162 file_uid = artifact.file_uid
163 response = self.app.get(route_path('download_file', fid=file_uid), status=200)
164 assert response.text == content
165
166 # log-in to new user and test download again
167 user = user_util.create_user(password='qweqwe')
168 self.log_user(user.username, 'qweqwe')
169 response = self.app.get(route_path('download_file', fid=file_uid), status=200)
170 assert response.text == content
171
172 def test_download_file_scoped_to_repo(self, user_util, create_artifact_factory):
173 user = self.log_user()
174 user_id = user['user_id']
175 content = 'HELLO MY NAME IS ARTIFACT !'
176
177 artifact = create_artifact_factory(user_id, content)
178 # bind to repo
179 repo = user_util.create_repo()
180 repo_id = repo.repo_id
181 artifact.scope_repo_id = repo_id
182 Session().add(artifact)
183 Session().commit()
184
185 file_uid = artifact.file_uid
186 response = self.app.get(route_path('download_file', fid=file_uid), status=200)
187 assert response.text == content
188
189 # log-in to new user and test download again
190 user = user_util.create_user(password='qweqwe')
191 self.log_user(user.username, 'qweqwe')
192 response = self.app.get(route_path('download_file', fid=file_uid), status=200)
193 assert response.text == content
194
195 # forbid user the rights to repo
196 repo = Repository.get(repo_id)
197 user_util.grant_user_permission_to_repo(repo, user, 'repository.none')
198 self.app.get(route_path('download_file', fid=file_uid), status=404)
199
200 def test_download_file_scoped_to_user(self, user_util, create_artifact_factory):
201 user = self.log_user()
202 user_id = user['user_id']
203 content = 'HELLO MY NAME IS ARTIFACT !'
204
205 artifact = create_artifact_factory(user_id, content)
206 # bind to user
207 user = user_util.create_user(password='qweqwe')
208
209 artifact.scope_user_id = user.user_id
210 Session().add(artifact)
211 Session().commit()
212
213 # artifact creator doesn't have access since it's bind to another user
214 file_uid = artifact.file_uid
215 self.app.get(route_path('download_file', fid=file_uid), status=404)
216
217 # log-in to new user and test download again, should be ok since we're bind to this artifact
218 self.log_user(user.username, 'qweqwe')
219 response = self.app.get(route_path('download_file', fid=file_uid), status=200)
220 assert response.text == content
221
222 def test_download_file_scoped_to_repo_with_bad_token(self, user_util, create_artifact_factory):
223 user_id = User.get_first_super_admin().user_id
224 content = 'HELLO MY NAME IS ARTIFACT !'
225
226 artifact = create_artifact_factory(user_id, content)
227 # bind to repo
228 repo = user_util.create_repo()
229 repo_id = repo.repo_id
230 artifact.scope_repo_id = repo_id
231 Session().add(artifact)
232 Session().commit()
233
234 file_uid = artifact.file_uid
235 self.app.get(route_path('download_file_by_token',
236 _auth_token='bogus', fid=file_uid), status=302)
237
238 def test_download_file_scoped_to_repo_with_token(self, user_util, create_artifact_factory):
239 user = User.get_first_super_admin()
240 AuthTokenModel().create(user, 'test artifact token',
241 role=AuthTokenModel.cls.ROLE_ARTIFACT_DOWNLOAD)
242
243 user = User.get_first_super_admin()
244 artifact_token = user.artifact_token
245
246 user_id = User.get_first_super_admin().user_id
247 content = 'HELLO MY NAME IS ARTIFACT !'
248
249 artifact = create_artifact_factory(user_id, content)
250 # bind to repo
251 repo = user_util.create_repo()
252 repo_id = repo.repo_id
253 artifact.scope_repo_id = repo_id
254 Session().add(artifact)
255 Session().commit()
256
257 file_uid = artifact.file_uid
258 response = self.app.get(
259 route_path('download_file_by_token',
260 _auth_token=artifact_token, fid=file_uid), status=200)
261 assert response.text == content
@@ -87,7 +87,7 b' class FileStoreView(BaseAppView):'
87 87 entry = FileStore.create(
88 88 file_uid=store_uid, filename=metadata["filename"],
89 89 file_hash=metadata["sha256"], file_size=metadata["size"],
90 file_description='upload attachment',
90 file_description=u'upload attachment',
91 91 check_acl=False, user_id=self._rhodecode_user.user_id
92 92 )
93 93 Session().add(entry)
@@ -116,6 +116,8 b' class FileStoreView(BaseAppView):'
116 116
117 117 # private upload for user
118 118 if db_obj.check_acl and db_obj.scope_user_id:
119 log.debug('Artifact: checking scope access for bound artifact user: `%s`',
120 db_obj.scope_user_id)
119 121 user = db_obj.user
120 122 if self._rhodecode_db_user.user_id != user.user_id:
121 123 log.warning('Access to file store object forbidden')
@@ -123,6 +125,8 b' class FileStoreView(BaseAppView):'
123 125
124 126 # scoped to repository permissions
125 127 if db_obj.check_acl and db_obj.scope_repo_id:
128 log.debug('Artifact: checking scope access for bound artifact repo: `%s`',
129 db_obj.scope_repo_id)
126 130 repo = db_obj.repo
127 131 perm_set = ['repository.read', 'repository.write', 'repository.admin']
128 132 has_perm = HasRepoPermissionAny(*perm_set)(repo.repo_name, 'FileStore check')
@@ -132,6 +136,8 b' class FileStoreView(BaseAppView):'
132 136
133 137 # scoped to repository group permissions
134 138 if db_obj.check_acl and db_obj.scope_repo_group_id:
139 log.debug('Artifact: checking scope access for bound artifact repo group: `%s`',
140 db_obj.scope_repo_group_id)
135 141 repo_group = db_obj.repo_group
136 142 perm_set = ['group.read', 'group.write', 'group.admin']
137 143 has_perm = HasRepoGroupPermissionAny(*perm_set)(repo_group.group_name, 'FileStore check')
@@ -152,7 +158,9 b' class FileStoreView(BaseAppView):'
152 158 log.debug('Requesting FID:%s from store %s', file_uid, self.storage)
153 159 return self._serve_file(file_uid)
154 160
161 # in addition to @LoginRequired ACL is checked by scopes
155 162 @LoginRequired(auth_token_access=[UserApiKeys.ROLE_ARTIFACT_DOWNLOAD])
163 @NotAnonymous()
156 164 @view_config(route_name='download_file_by_token')
157 165 def download_file_by_token(self):
158 166 """
General Comments 0
You need to be logged in to leave comments. Login now