From be17ae68ee70c4be7db6a1c1a9c9849f08cf667d Mon Sep 17 00:00:00 2001 From: 0mar Date: Wed, 7 Apr 2021 08:38:05 +0200 Subject: [PATCH 1/4] Upgraded to log warning, resolved comment --- jupyterhub/app.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/jupyterhub/app.py b/jupyterhub/app.py index f4f7440d..5cdb5aa6 100644 --- a/jupyterhub/app.py +++ b/jupyterhub/app.py @@ -1890,7 +1890,7 @@ class JupyterHub(Application): # make sure that on no admin situation, all roles are reset admin_role = orm.Role.find(db, name='admin') if not admin_role.users: - app_log.info( + app_log.warning( "No admin users found; assuming hub upgrade. Initializing default roles for all entities" ) # make sure all users, services and tokens have at least one role (update with default) From 4577266d95bc492c7e29e4a5638c3a367b88b349 Mon Sep 17 00:00:00 2001 From: 0mar Date: Sat, 10 Apr 2021 16:24:20 +0200 Subject: [PATCH 2/4] Refactored scope test suite with proper fixtures and teardowns --- jupyterhub/tests/test_scopes.py | 359 ++++++++++++++++---------------- 1 file changed, 175 insertions(+), 184 deletions(-) diff --git a/jupyterhub/tests/test_scopes.py b/jupyterhub/tests/test_scopes.py index b97b4728..496d053a 100644 --- a/jupyterhub/tests/test_scopes.py +++ b/jupyterhub/tests/test_scopes.py @@ -89,6 +89,10 @@ class MockAPIHandler: self.request = mock.Mock(spec=HTTPServerRequest) self.request.path = '/path' + def set_scopes(self, *scopes): + self.raw_scopes = set(scopes) + self.parsed_scopes = parse_scopes(self.raw_scopes) + @needs_scope('users') def user_thing(self, user_name): return True @@ -116,6 +120,12 @@ class MockAPIHandler: return True +@pytest.fixture +def mock_handler(): + obj = MockAPIHandler() + return obj + + @mark.parametrize( "scopes, method, arguments, is_allowed", [ @@ -169,12 +179,11 @@ class MockAPIHandler: (['users!user=gob'], 'other_thing', ('maeby',), True), ], ) -def test_scope_method_access(scopes, method, arguments, is_allowed): - obj = MockAPIHandler() - obj.current_user = mock.Mock(name=arguments[0]) - obj.raw_scopes = set(scopes) - obj.parsed_scopes = parse_scopes(obj.raw_scopes) - api_call = getattr(obj, method) +def test_scope_method_access(mock_handler, scopes, method, arguments, is_allowed): + mock_handler = MockAPIHandler() + mock_handler.current_user = mock.Mock(name=arguments[0]) + mock_handler.set_scopes(*scopes) + api_call = getattr(mock_handler, method) if is_allowed: assert api_call(*arguments) else: @@ -182,31 +191,18 @@ def test_scope_method_access(scopes, method, arguments, is_allowed): api_call(*arguments) -def test_double_scoped_method_succeeds(): - obj = MockAPIHandler() - obj.current_user = mock.Mock(name='lucille') - obj.raw_scopes = {'users', 'read:services'} - obj.parsed_scopes = parse_scopes(obj.raw_scopes) - assert obj.secret_thing() +def test_double_scoped_method_succeeds(mock_handler): + mock_handler.current_user = mock.Mock(name='lucille') + mock_handler.set_scopes('users', 'read:services') + mock_handler.parsed_scopes = parse_scopes(mock_handler.raw_scopes) + assert mock_handler.secret_thing() -def test_double_scoped_method_denials(): - obj = MockAPIHandler() - obj.current_user = mock.Mock(name='lucille2') - obj.raw_scopes = {'users', 'read:groups'} - obj.parsed_scopes = parse_scopes(obj.raw_scopes) +def test_double_scoped_method_denials(mock_handler): + mock_handler.current_user = mock.Mock(name='lucille2') + mock_handler.set_scopes('users', 'read:groups') with pytest.raises(web.HTTPError): - obj.secret_thing() - - -def generate_test_role(user_name, scopes, role_name='test'): - role = { - 'name': role_name, - 'description': '', - 'users': [user_name], - 'scopes': scopes, - } - return role + mock_handler.secret_thing() @mark.parametrize( @@ -247,6 +243,8 @@ async def test_expand_groups(app, user_name, in_group, status_code): app, 'users', user_name, headers=auth_header(app.db, user_name) ) assert r.status_code == status_code + app.db.delete(group) + app.db.commit() async def test_by_fake_user(app): @@ -262,79 +260,121 @@ async def test_by_fake_user(app): err_message = "No access to resources or resources not found" -async def test_request_fake_user(app): - user_name = 'buster' - fake_user = 'annyong' - user = add_user(app.db, name=user_name) - test_role = generate_test_role(user_name, ['read:users!group=stuff']) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') +@pytest.fixture +def temp_role(app): + """Generate a temporary role with certain scopes. + Convenience function that takes care of db stuff and teardown""" + temp_roles = [] + index = [1] + + def temp_role_creator(scopes, role_name=None): + if not role_name: + role_name = f'test_role_{index[0]}' + index[0] += 1 + role = orm.Role(name=role_name, scopes=list(scopes)) + temp_roles.append(role) + app.db.add(role) + app.db.commit() + return role + + yield temp_role_creator + for role in temp_roles: + app.db.delete(role) app.db.commit() + + +@pytest.fixture +def user_with_scopes(app, temp_role): + """Generate a temporary user with specific scopes. + Takes care of db stuff and teardown""" + temp_users = [] + index = [1] + role_function = temp_role + + def temp_user_creator(*scopes): + name = f"temporary_user_{index[0]}" + index[0] += 1 + role = role_function(scopes) + orm_user = orm.User(name=name) + app.db.add(orm_user) + app.db.commit() + temp_users.append(orm_user) + roles.update_roles(app.db, orm_user, roles=[role.name]) + return app.users[orm_user.id] + + yield temp_user_creator + for user in temp_users: + app.users.delete(user) + + +@pytest.fixture +def service_with_scopes(app, temp_role): + """Generate a temporary service with specific scopes. + Takes care of db stuff and teardown""" + temp_service = [] + index = [1] + role_function = temp_role + + def temp_service_creator(*scopes): + name = f"temporary_service_{index[0]}" + index[0] += 1 + role = role_function(scopes) + app.services.append({'name': name}) + app.init_services() + orm_service = orm.Service.find(app.db, name) + app.db.commit() + roles.update_roles(app.db, orm_service, roles=[role.name]) + return orm_service + + yield temp_service_creator + for service in temp_service: + app.db.delete(service) + app.db.commit() + + +async def test_request_fake_user(app, user_with_scopes): + fake_user = 'annyong' + user = user_with_scopes('read:users!group=stuff') r = await api_request( - app, 'users', fake_user, headers=auth_header(app.db, user_name) + app, 'users', fake_user, headers=auth_header(app.db, user.name) ) assert r.status_code == 404 # Consistency between no user and user not accessible assert r.json()['message'] == err_message -async def test_refuse_exceeding_token_permissions(app): - user_name = 'abed' - user = add_user(app.db, name=user_name) - add_user(app.db, name='user') - api_token = user.new_api_token() - exceeding_role = generate_test_role(user_name, ['read:users'], 'exceeding_role') - roles.create_role(app.db, exceeding_role) - roles.grant_role(app.db, entity=user.api_tokens[0], rolename='exceeding_role') - app.db.commit() - headers = {'Authorization': 'token %s' % api_token} - r = await api_request(app, 'users', headers=headers) - assert r.status_code == 200 - result_names = {user['name'] for user in r.json()} - assert result_names == {user_name} +async def test_refuse_exceeding_token_permissions(app, user_with_scopes, temp_role): + user = user_with_scopes('self') + user.new_api_token() + temp_role(['admin:users'], 'exceeding_role') + with pytest.raises(ValueError): + roles.update_roles(app.db, entity=user.api_tokens[0], roles=['exceeding_role']) -async def test_exceeding_user_permissions(app): - user_name = 'abed' - user = add_user(app.db, name=user_name) - add_user(app.db, name='user') +async def test_exceeding_user_permissions(app, user_with_scopes, temp_role): + user = user_with_scopes('read:users:groups') api_token = user.new_api_token() orm_api_token = orm.APIToken.find(app.db, token=api_token) - reader_role = generate_test_role(user_name, ['read:users'], 'reader_role') - subreader_role = generate_test_role( - user_name, ['read:users:groups'], 'subreader_role' - ) - roles.create_role(app.db, reader_role) - roles.create_role(app.db, subreader_role) - app.db.commit() - roles.update_roles(app.db, user, roles=['reader_role']) - roles.update_roles(app.db, orm_api_token, roles=['subreader_role']) - orm_api_token.roles.remove(orm.Role.find(app.db, name='token')) - app.db.commit() - + temp_role(['read:users'], 'reader_role') + roles.grant_role(app.db, orm_api_token, rolename='reader_role') headers = {'Authorization': 'token %s' % api_token} r = await api_request(app, 'users', headers=headers) assert r.status_code == 200 keys = {key for user in r.json() for key in user.keys()} assert 'groups' in keys assert 'last_activity' not in keys - roles.strip_role(app.db, user, 'reader_role') -async def test_user_service_separation(app, mockservice_url): +async def test_user_service_separation(app, mockservice_url, temp_role): name = mockservice_url.name user = add_user(app.db, name=name) - reader_role = generate_test_role(name, ['read:users'], 'reader_role') - subreader_role = generate_test_role(name, ['read:users:groups'], 'subreader_role') - roles.create_role(app.db, reader_role) - roles.create_role(app.db, subreader_role) - app.db.commit() + temp_role(['read:users'], 'reader_role') + temp_role(['read:users:groups'], 'subreader_role') roles.update_roles(app.db, user, roles=['subreader_role']) roles.update_roles(app.db, mockservice_url.orm, roles=['reader_role']) user.roles.remove(orm.Role.find(app.db, name='user')) api_token = user.new_api_token() - app.db.commit() headers = {'Authorization': 'token %s' % api_token} r = await api_request(app, 'users', headers=headers) assert r.status_code == 200 @@ -343,33 +383,22 @@ async def test_user_service_separation(app, mockservice_url): assert 'last_activity' not in keys -async def test_request_user_outside_group(app): - user_name = 'buster' - fake_user = 'hello' - user = add_user(app.db, name=user_name) - add_user(app.db, name=fake_user) - test_role = generate_test_role(user_name, ['read:users!group=stuff']) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') - roles.strip_role(app.db, entity=user, rolename='user') - app.db.commit() +async def test_request_user_outside_group(app, user_with_scopes): + outside_user = 'hello' + user = user_with_scopes('read:users!group=stuff') + add_user(app.db, name=outside_user) r = await api_request( - app, 'users', fake_user, headers=auth_header(app.db, user_name) + app, 'users', outside_user, headers=auth_header(app.db, user.name) ) assert r.status_code == 404 # Consistency between no user and user not accessible assert r.json()['message'] == err_message -async def test_user_filter(app): - user_name = 'rita' - user = add_user(app.db, name=user_name) - app.db.commit() - scopes = ['read:users!user=lindsay', 'read:users!user=gob', 'read:users!user=oscar'] - test_role = generate_test_role(user, scopes) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') - roles.strip_role(app.db, entity=user, rolename='user') +async def test_user_filter(app, user_with_scopes): + user = user_with_scopes( + 'read:users!user=lindsay', 'read:users!user=gob', 'read:users!user=oscar' + ) name_in_scope = {'lindsay', 'oscar', 'gob'} outside_scope = {'maeby', 'marta'} group_name = 'bluth' @@ -378,17 +407,19 @@ async def test_user_filter(app): group = orm.Group(name=group_name) app.db.add(group) for name in name_in_scope | outside_scope: - user = add_user(app.db, name=name) + group_user = add_user(app.db, name=name) if name not in group.users: - group.users.append(user) + group.users.append(group_user) app.db.commit() - r = await api_request(app, 'users', headers=auth_header(app.db, user_name)) + r = await api_request(app, 'users', headers=auth_header(app.db, user.name)) assert r.status_code == 200 result_names = {user['name'] for user in r.json()} assert result_names == name_in_scope + app.db.delete(group) + app.db.commit() -async def test_service_filter(app): +async def test_service_filter(app, user_with_scopes): services = [ {'name': 'cull_idle', 'api_token': 'some-token'}, {'name': 'user_service', 'api_token': 'some-other-token'}, @@ -396,124 +427,88 @@ async def test_service_filter(app): for service in services: app.services.append(service) app.init_services() - user_name = 'buster' - user = add_user(app.db, name=user_name) - app.db.commit() - test_role = generate_test_role(user, ['read:services!service=cull_idle']) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') - r = await api_request(app, 'services', headers=auth_header(app.db, user_name)) + user = user_with_scopes('read:services!service=cull_idle') + r = await api_request(app, 'services', headers=auth_header(app.db, user.name)) assert r.status_code == 200 service_names = set(r.json().keys()) assert service_names == {'cull_idle'} -async def test_user_filter_with_group(app): - # Move role setup to setup method? - user_name = 'sally' - user = add_user(app.db, name=user_name) - external_user_name = 'britta' - add_user(app.db, name=external_user_name) - test_role = generate_test_role(user_name, ['read:users!group=sitwell']) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') - - name_set = {'sally', 'stan'} +async def test_user_filter_with_group(app, user_with_scopes): group_name = 'sitwell' + user1 = user_with_scopes(f'read:users!group={group_name}') + user2 = user_with_scopes('self') + external_user = user_with_scopes('self') + name_set = {user1.name, user2.name} group = orm.Group.find(app.db, name=group_name) if not group: group = orm.Group(name=group_name) app.db.add(group) - for name in name_set: - user = add_user(app.db, name=name) - if name not in group.users: - group.users.append(user) + for user in {user1, user2}: + group.users.append(user) app.db.commit() - r = await api_request(app, 'users', headers=auth_header(app.db, user_name)) + r = await api_request(app, 'users', headers=auth_header(app.db, user1.name)) assert r.status_code == 200 result_names = {user['name'] for user in r.json()} assert result_names == name_set - assert external_user_name not in result_names + assert external_user.name not in result_names + app.db.delete(group) + app.db.commit() -async def test_group_scope_filter(app): - user_name = 'rollerblade' - user = add_user(app.db, name=user_name) - scopes = ['read:groups!group=sitwell', 'read:groups!group=bluth'] - test_role = generate_test_role(user_name, scopes) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') - - group_set = {'sitwell', 'bluth', 'austero'} - for group_name in group_set: +async def test_group_scope_filter(app, user_with_scopes): + in_groups = {'sitwell', 'bluth'} + out_groups = {'austero'} + user = user_with_scopes(*(f'read:groups!group={group}' for group in in_groups)) + for group_name in in_groups | out_groups: group = orm.Group.find(app.db, name=group_name) if not group: group = orm.Group(name=group_name) app.db.add(group) app.db.commit() - r = await api_request(app, 'groups', headers=auth_header(app.db, user_name)) + r = await api_request(app, 'groups', headers=auth_header(app.db, user.name)) assert r.status_code == 200 result_names = {user['name'] for user in r.json()} - assert result_names == {'sitwell', 'bluth'} - - -async def test_vertical_filter(app): - user_name = 'lindsey' - user = add_user(app.db, name=user_name) - test_role = generate_test_role(user_name, ['read:users:name']) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') - roles.strip_role(app.db, entity=user, rolename='user') + assert result_names == in_groups + for group_name in in_groups | out_groups: + group = orm.Group.find(app.db, name=group_name) + app.db.delete(group) app.db.commit() - r = await api_request(app, 'users', headers=auth_header(app.db, user_name)) + +async def test_vertical_filter(app, user_with_scopes): + user = user_with_scopes('read:users:name') + r = await api_request(app, 'users', headers=auth_header(app.db, user.name)) assert r.status_code == 200 allowed_keys = {'name', 'kind'} assert set([key for user in r.json() for key in user.keys()]) == allowed_keys -async def test_stacked_vertical_filter(app): - user_name = 'user' - user = add_user(app.db, name=user_name) - test_role = generate_test_role( - user_name, ['read:users:activity', 'read:users:servers'] - ) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') - roles.strip_role(app.db, entity=user, rolename='user') - app.db.commit() - - r = await api_request(app, 'users', headers=auth_header(app.db, user_name)) +async def test_stacked_vertical_filter(app, user_with_scopes): + user = user_with_scopes('read:users:activity', 'read:users:servers') + r = await api_request(app, 'users', headers=auth_header(app.db, user.name)) assert r.status_code == 200 allowed_keys = {'name', 'kind', 'servers', 'last_activity'} result_model = set([key for user in r.json() for key in user.keys()]) assert result_model == allowed_keys -async def test_cross_filter(app): - user_name = 'abed' - user = add_user(app.db, name=user_name) - test_role = generate_test_role( - user_name, ['read:users:activity', 'read:users!user=abed'] - ) - roles.create_role(app.db, test_role) - roles.grant_role(app.db, entity=user, rolename='test') - roles.strip_role(app.db, entity=user, rolename='user') - app.db.commit() +async def test_cross_filter(app, user_with_scopes): + user = user_with_scopes('read:users:activity', 'self') new_users = {'britta', 'jeff', 'annie'} for new_user_name in new_users: add_user(app.db, name=new_user_name) app.db.commit() - r = await api_request(app, 'users', headers=auth_header(app.db, user_name)) + r = await api_request(app, 'users', headers=auth_header(app.db, user.name)) assert r.status_code == 200 restricted_keys = {'name', 'kind', 'last_activity'} key_in_full_model = 'created' - for user in r.json(): - if user['name'] == user_name: - assert key_in_full_model in user + for model_user in r.json(): + if model_user['name'] == user.name: + assert key_in_full_model in model_user else: - assert set(user.keys()) == restricted_keys + assert set(model_user.keys()) == restricted_keys @mark.parametrize( @@ -523,13 +518,13 @@ async def test_cross_filter(app): ('services', False), ], ) -async def test_metascope_self_expansion(app, kind, has_user_scopes): - Class = orm.get_class(kind) - orm_obj = Class(name=f'test_{kind}') - app.db.add(orm_obj) - app.db.commit() - test_role = orm.Role(name='test_role', scopes=['self']) - orm_obj.roles.append(test_role) +async def test_metascope_self_expansion( + app, kind, has_user_scopes, user_with_scopes, service_with_scopes +): + if kind == 'users': + orm_obj = user_with_scopes('self') + else: + orm_obj = service_with_scopes('self') # test expansion of user/service scopes scopes = roles.expand_roles_to_scopes(orm_obj) assert bool(scopes) == has_user_scopes @@ -538,14 +533,10 @@ async def test_metascope_self_expansion(app, kind, has_user_scopes): orm_obj.new_api_token() token_scopes = get_scopes_for(orm_obj.api_tokens[0]) assert bool(token_scopes) == has_user_scopes - app.db.delete(orm_obj) - app.db.delete(test_role) - app.db.commit() -async def test_metascope_all_expansion(app): - user = add_user(app.db, name='user') - scope_set = {scope for role in user.roles for scope in role.scopes} +async def test_metascope_all_expansion(app, user_with_scopes): + user = user_with_scopes('self') user.new_api_token() token = user.api_tokens[0] # Check 'all' expansion From a7f2247331809ca14543c893d7e49a7955d35ec0 Mon Sep 17 00:00:00 2001 From: 0mar Date: Mon, 12 Apr 2021 16:49:29 +0200 Subject: [PATCH 3/4] Minor fixes --- jupyterhub/roles.py | 2 +- jupyterhub/tests/test_scopes.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/jupyterhub/roles.py b/jupyterhub/roles.py index 81ba6425..e4bc2e1e 100644 --- a/jupyterhub/roles.py +++ b/jupyterhub/roles.py @@ -371,7 +371,7 @@ def _token_allowed_role(db, token, role): raw_owner_scopes = { scope.split('!', 1)[0] if '!' in scope else scope for scope in owner_scopes } - if (raw_extra_scopes).issubset(raw_owner_scopes): + if raw_extra_scopes.issubset(raw_owner_scopes): return True else: return False diff --git a/jupyterhub/tests/test_scopes.py b/jupyterhub/tests/test_scopes.py index 496d053a..2799af44 100644 --- a/jupyterhub/tests/test_scopes.py +++ b/jupyterhub/tests/test_scopes.py @@ -613,7 +613,7 @@ async def test_server_state_access( service.roles.append(role) app.db.commit() api_token = service.new_api_token() - app.init_roles() + await app.init_roles() headers = {'Authorization': 'token %s' % api_token} r = await api_request(app, 'users', username, headers=headers) r.raise_for_status() From 834694ca7e3413cf71404b3642bbd255aec455db Mon Sep 17 00:00:00 2001 From: 0mar Date: Tue, 13 Apr 2021 18:08:51 +0200 Subject: [PATCH 4/4] Refactored names and suggested fixes --- jupyterhub/tests/test_scopes.py | 118 +++++++++++++++++--------------- 1 file changed, 62 insertions(+), 56 deletions(-) diff --git a/jupyterhub/tests/test_scopes.py b/jupyterhub/tests/test_scopes.py index 2799af44..c0f4269d 100644 --- a/jupyterhub/tests/test_scopes.py +++ b/jupyterhub/tests/test_scopes.py @@ -180,7 +180,6 @@ def mock_handler(): ], ) def test_scope_method_access(mock_handler, scopes, method, arguments, is_allowed): - mock_handler = MockAPIHandler() mock_handler.current_user = mock.Mock(name=arguments[0]) mock_handler.set_scopes(*scopes) api_call = getattr(mock_handler, method) @@ -235,7 +234,6 @@ async def test_expand_groups(app, user_name, in_group, status_code): app.db.add(group) if in_group and user not in group.users: group.users.append(user) - kind = 'users' roles.update_roles(app.db, user, roles=['test']) roles.strip_role(app.db, user, 'user') app.db.commit() @@ -261,21 +259,21 @@ err_message = "No access to resources or resources not found" @pytest.fixture -def temp_role(app): +def create_temp_role(app): """Generate a temporary role with certain scopes. - Convenience function that takes care of db stuff and teardown""" + Convenience function that provides setup, database handling and teardown""" temp_roles = [] index = [1] def temp_role_creator(scopes, role_name=None): if not role_name: - role_name = f'test_role_{index[0]}' + role_name = f'temp_role_{index[0]}' index[0] += 1 - role = orm.Role(name=role_name, scopes=list(scopes)) - temp_roles.append(role) - app.db.add(role) + temp_role = orm.Role(name=role_name, scopes=list(scopes)) + temp_roles.append(temp_role) + app.db.add(temp_role) app.db.commit() - return role + return temp_role yield temp_role_creator for role in temp_roles: @@ -284,17 +282,18 @@ def temp_role(app): @pytest.fixture -def user_with_scopes(app, temp_role): +def create_user_with_scopes(app, create_temp_role): """Generate a temporary user with specific scopes. - Takes care of db stuff and teardown""" + Convenience function that provides setup, database handling and teardown""" temp_users = [] - index = [1] - role_function = temp_role + counter = 0 + get_role = create_temp_role def temp_user_creator(*scopes): - name = f"temporary_user_{index[0]}" - index[0] += 1 - role = role_function(scopes) + nonlocal counter + counter += 1 + name = f"temp_user_{counter}" + role = get_role(scopes) orm_user = orm.User(name=name) app.db.add(orm_user) app.db.commit() @@ -308,16 +307,17 @@ def user_with_scopes(app, temp_role): @pytest.fixture -def service_with_scopes(app, temp_role): +def create_service_with_scopes(app, create_temp_role): """Generate a temporary service with specific scopes. - Takes care of db stuff and teardown""" + Convenience function that provides setup, database handling and teardown""" temp_service = [] - index = [1] - role_function = temp_role + counter = 0 + role_function = create_temp_role def temp_service_creator(*scopes): - name = f"temporary_service_{index[0]}" - index[0] += 1 + nonlocal counter + counter += 1 + name = f"temp_service_{counter}" role = role_function(scopes) app.services.append({'name': name}) app.init_services() @@ -332,9 +332,9 @@ def service_with_scopes(app, temp_role): app.db.commit() -async def test_request_fake_user(app, user_with_scopes): +async def test_request_fake_user(app, create_user_with_scopes): fake_user = 'annyong' - user = user_with_scopes('read:users!group=stuff') + user = create_user_with_scopes('read:users!group=stuff') r = await api_request( app, 'users', fake_user, headers=auth_header(app.db, user.name) ) @@ -343,19 +343,23 @@ async def test_request_fake_user(app, user_with_scopes): assert r.json()['message'] == err_message -async def test_refuse_exceeding_token_permissions(app, user_with_scopes, temp_role): - user = user_with_scopes('self') +async def test_refuse_exceeding_token_permissions( + app, create_user_with_scopes, create_temp_role +): + user = create_user_with_scopes('self') user.new_api_token() - temp_role(['admin:users'], 'exceeding_role') + create_temp_role(['admin:users'], 'exceeding_role') with pytest.raises(ValueError): roles.update_roles(app.db, entity=user.api_tokens[0], roles=['exceeding_role']) -async def test_exceeding_user_permissions(app, user_with_scopes, temp_role): - user = user_with_scopes('read:users:groups') +async def test_exceeding_user_permissions( + app, create_user_with_scopes, create_temp_role +): + user = create_user_with_scopes('read:users:groups') api_token = user.new_api_token() orm_api_token = orm.APIToken.find(app.db, token=api_token) - temp_role(['read:users'], 'reader_role') + create_temp_role(['read:users'], 'reader_role') roles.grant_role(app.db, orm_api_token, rolename='reader_role') headers = {'Authorization': 'token %s' % api_token} r = await api_request(app, 'users', headers=headers) @@ -365,12 +369,12 @@ async def test_exceeding_user_permissions(app, user_with_scopes, temp_role): assert 'last_activity' not in keys -async def test_user_service_separation(app, mockservice_url, temp_role): +async def test_user_service_separation(app, mockservice_url, create_temp_role): name = mockservice_url.name user = add_user(app.db, name=name) - temp_role(['read:users'], 'reader_role') - temp_role(['read:users:groups'], 'subreader_role') + create_temp_role(['read:users'], 'reader_role') + create_temp_role(['read:users:groups'], 'subreader_role') roles.update_roles(app.db, user, roles=['subreader_role']) roles.update_roles(app.db, mockservice_url.orm, roles=['reader_role']) user.roles.remove(orm.Role.find(app.db, name='user')) @@ -383,9 +387,9 @@ async def test_user_service_separation(app, mockservice_url, temp_role): assert 'last_activity' not in keys -async def test_request_user_outside_group(app, user_with_scopes): +async def test_request_user_outside_group(app, create_user_with_scopes): outside_user = 'hello' - user = user_with_scopes('read:users!group=stuff') + user = create_user_with_scopes('read:users!group=stuff') add_user(app.db, name=outside_user) r = await api_request( app, 'users', outside_user, headers=auth_header(app.db, user.name) @@ -395,8 +399,8 @@ async def test_request_user_outside_group(app, user_with_scopes): assert r.json()['message'] == err_message -async def test_user_filter(app, user_with_scopes): - user = user_with_scopes( +async def test_user_filter(app, create_user_with_scopes): + user = create_user_with_scopes( 'read:users!user=lindsay', 'read:users!user=gob', 'read:users!user=oscar' ) name_in_scope = {'lindsay', 'oscar', 'gob'} @@ -419,7 +423,7 @@ async def test_user_filter(app, user_with_scopes): app.db.commit() -async def test_service_filter(app, user_with_scopes): +async def test_service_filter(app, create_user_with_scopes): services = [ {'name': 'cull_idle', 'api_token': 'some-token'}, {'name': 'user_service', 'api_token': 'some-other-token'}, @@ -427,18 +431,18 @@ async def test_service_filter(app, user_with_scopes): for service in services: app.services.append(service) app.init_services() - user = user_with_scopes('read:services!service=cull_idle') + user = create_user_with_scopes('read:services!service=cull_idle') r = await api_request(app, 'services', headers=auth_header(app.db, user.name)) assert r.status_code == 200 service_names = set(r.json().keys()) assert service_names == {'cull_idle'} -async def test_user_filter_with_group(app, user_with_scopes): +async def test_user_filter_with_group(app, create_user_with_scopes): group_name = 'sitwell' - user1 = user_with_scopes(f'read:users!group={group_name}') - user2 = user_with_scopes('self') - external_user = user_with_scopes('self') + user1 = create_user_with_scopes(f'read:users!group={group_name}') + user2 = create_user_with_scopes('self') + external_user = create_user_with_scopes('self') name_set = {user1.name, user2.name} group = orm.Group.find(app.db, name=group_name) if not group: @@ -457,10 +461,12 @@ async def test_user_filter_with_group(app, user_with_scopes): app.db.commit() -async def test_group_scope_filter(app, user_with_scopes): +async def test_group_scope_filter(app, create_user_with_scopes): in_groups = {'sitwell', 'bluth'} out_groups = {'austero'} - user = user_with_scopes(*(f'read:groups!group={group}' for group in in_groups)) + user = create_user_with_scopes( + *(f'read:groups!group={group}' for group in in_groups) + ) for group_name in in_groups | out_groups: group = orm.Group.find(app.db, name=group_name) if not group: @@ -477,16 +483,16 @@ async def test_group_scope_filter(app, user_with_scopes): app.db.commit() -async def test_vertical_filter(app, user_with_scopes): - user = user_with_scopes('read:users:name') +async def test_vertical_filter(app, create_user_with_scopes): + user = create_user_with_scopes('read:users:name') r = await api_request(app, 'users', headers=auth_header(app.db, user.name)) assert r.status_code == 200 allowed_keys = {'name', 'kind'} assert set([key for user in r.json() for key in user.keys()]) == allowed_keys -async def test_stacked_vertical_filter(app, user_with_scopes): - user = user_with_scopes('read:users:activity', 'read:users:servers') +async def test_stacked_vertical_filter(app, create_user_with_scopes): + user = create_user_with_scopes('read:users:activity', 'read:users:servers') r = await api_request(app, 'users', headers=auth_header(app.db, user.name)) assert r.status_code == 200 allowed_keys = {'name', 'kind', 'servers', 'last_activity'} @@ -494,8 +500,8 @@ async def test_stacked_vertical_filter(app, user_with_scopes): assert result_model == allowed_keys -async def test_cross_filter(app, user_with_scopes): - user = user_with_scopes('read:users:activity', 'self') +async def test_cross_filter(app, create_user_with_scopes): + user = create_user_with_scopes('read:users:activity', 'self') new_users = {'britta', 'jeff', 'annie'} for new_user_name in new_users: add_user(app.db, name=new_user_name) @@ -519,12 +525,12 @@ async def test_cross_filter(app, user_with_scopes): ], ) async def test_metascope_self_expansion( - app, kind, has_user_scopes, user_with_scopes, service_with_scopes + app, kind, has_user_scopes, create_user_with_scopes, create_service_with_scopes ): if kind == 'users': - orm_obj = user_with_scopes('self') + orm_obj = create_user_with_scopes('self') else: - orm_obj = service_with_scopes('self') + orm_obj = create_service_with_scopes('self') # test expansion of user/service scopes scopes = roles.expand_roles_to_scopes(orm_obj) assert bool(scopes) == has_user_scopes @@ -535,8 +541,8 @@ async def test_metascope_self_expansion( assert bool(token_scopes) == has_user_scopes -async def test_metascope_all_expansion(app, user_with_scopes): - user = user_with_scopes('self') +async def test_metascope_all_expansion(app, create_user_with_scopes): + user = create_user_with_scopes('self') user.new_api_token() token = user.api_tokens[0] # Check 'all' expansion