fixed: Permission error occured when accessing the JobCommits #4434

[k1won]

Motivation and context

Resolves #4434

How has this been tested?

Manual test

Checklist

• I submit my changes into the develop branch
• I have added a description of my changes into CHANGELOG file
• I have updated the documentation accordingly
• I have added tests to cover my changes
• I have linked related issues (read github docs)
• I have increased versions of npm packages if it is necessary (cvat-canvas,
  cvat-core, cvat-data and cvat-ui)

License

• I submit my code changes under the same MIT License that covers the project.
  Feel free to contact the maintainers if that's a concern.
• I have updated the license header for each file (see an example below)

# Copyright (C) 2022 Intel Corporation
#
# SPDX-License-Identifier: MIT

[nmanovic]

@k1won , need to update the file https://github.com/openvinotoolkit/cvat/blob/develop/cvat/apps/iam/rules/jobs.csv

[nmanovic]

Need to generate new tests: https://github.com/openvinotoolkit/cvat/blob/develop/cvat/apps/iam/rules/jobs_test.gen.rego. You can find the python script to generate it at the end of the file (as comments).

[nmanovic]

# jobs_test.gen.rego.py
# # Copyright (C) 2021 Intel Corporation
# #
# # SPDX-License-Identifier: MIT
#
# # Copyright (C) 2021 Intel Corporation
# #
# # SPDX-License-Identifier: MIT
#
# import csv
# import json
# import random
# import sys
# import os
# from itertools import product
#
#
# NAME = 'jobs'
#
# def read_rules(name):
#     rules = []
#     with open(os.path.join(sys.argv[1], f'{name}.csv')) as f:
#         reader = csv.DictReader(f)
#         for row in reader:
#             row = {k.lower():v.lower().replace('n/a','na') for k,v in row.items()}
#             row['limit'] = row['limit'].replace('none', 'None')
#             found = False
#             for col,val in row.items():
#                 if col in ["limit", "method", "url", "resource"]:
#                     continue
#                 complex_val = [v.strip() for v in val.split(',')]
#                 if len(complex_val) > 1:
#                     found = True
#                     for item in complex_val:
#                         new_row = row.copy()
#                         new_row[col] = item
#                         rules.append(new_row)
#             if not found:
#                 rules.append(row)
#
#     return rules
#
# simple_rules = read_rules(NAME)
#
# SCOPES = list({rule['scope'] for rule in simple_rules})
# CONTEXTS = ['sandbox', 'organization']
# OWNERSHIPS = ['project:owner', 'project:assignee', 'task:owner', 'task:assignee', 'assignee', 'none']
# GROUPS = ['admin', 'business', 'user', 'worker', 'none']
# ORG_ROLES = ['owner', 'maintainer', 'supervisor', 'worker', None]
# SAME_ORG = [True, False]
#
# def RESOURCES(scope):
#     if scope == 'list':
#         return [None]
#     else:
#         return [{
#             "id": random.randrange(300, 400),
#             "assignee": { "id": random.randrange(500, 600) },
#             "organization": {
#                 "id": random.randrange(600, 700)
#             },
#             "project": {
#                 "id": random.randrange(300, 400),
#                 "owner": { "id": random.randrange(700, 800) },
#                 "assignee": { "id": random.randrange(800, 900) }
#             },
#             "task": {
#                 "id": random.randrange(300, 400),
#                 "owner": { "id": random.randrange(900, 1000) },
#                 "assignee": { "id": random.randrange(1000, 1100) }
#             }
#         }]
#
# def is_same_org(org1, org2):
#     if org1 != None and org2 != None:
#         return org1['id'] == org2['id']
#     elif org1 == None and org2 == None:
#         return True
#     else:
#         return False
#
# def eval_rule(scope, context, ownership, privilege, membership, data):
#     if privilege == 'admin':
#         return True
#
#     rules = list(filter(lambda r: scope == r['scope'], simple_rules))
#     rules = list(filter(lambda r: r['context'] == 'na' or context == r['context'], rules))
#     rules = list(filter(lambda r: r['ownership'] == 'na' or ownership == r['ownership'], rules))
#     rules = list(filter(lambda r: r['membership'] == 'na' or
#         ORG_ROLES.index(membership) <= ORG_ROLES.index(r['membership']), rules))
#     rules = list(filter(lambda r: GROUPS.index(privilege) <= GROUPS.index(r['privilege']), rules))
#     resource = data['resource']
#     rules = list(filter(lambda r: not r['limit'] or eval(r['limit'], {'resource': resource}), rules))
#     if not is_same_org(data['auth']['organization'], data['resource']['organization']) and context != 'sandbox':
#         return False
#
#     return bool(rules)
#
# def get_data(scope, context, ownership, privilege, membership, resource, same_org):
#     data = {
#         "scope": scope,
#         "auth": {
#             "user": { "id": random.randrange(0,100), "privilege": privilege },
#             "organization": {
#                 "id": random.randrange(100,200),
#                 "owner": { "id": random.randrange(200, 300) },
#                 "user": { "role": membership }
#             } if context == 'organization' else None
#         },
#         "resource": resource
#     }
#
#     user_id = data['auth']['user']['id']
#     if context == 'organization':
#         org_id = data['auth']['organization']['id']
#         if data['auth']['organization']['user']['role'] == 'owner':
#             data['auth']['organization']['owner']['id'] = user_id
#
#         if same_org:
#             data['resource']['organization']['id'] = org_id
#
#     if ownership == 'assignee':
#         data['resource']['assignee']['id'] = user_id
#
#     if ownership == 'project:owner':
#         data['resource']['project']['owner']['id'] = user_id
#
#     if ownership == 'project:assignee':
#         data['resource']['project']['assignee']['id'] = user_id
#
#     if ownership == 'task:owner':
#         data['resource']['task']['owner']['id'] = user_id
#
#     if ownership == 'task:assignee':
#         data['resource']['task']['assignee']['id'] = user_id
#
#     return data
#
# def _get_name(prefix, **kwargs):
#     name = prefix
#     for k,v in kwargs.items():
#         if k == 'resource':
#             continue
#         prefix = '_' + str(k)
#         if isinstance(v, dict):
#             if 'id' in v:
#                 v = v.copy()
#                 v.pop('id')
#             if v:
#                 name += _get_name(prefix, **v)
#         else:
#             name += ''.join(map(lambda c: c if c.isalnum() else {'@':'_IN_'}.get(c, '_'),
#                 f'{prefix}_{str(v).upper()}'))
#
#     return name
#
# def get_name(scope, context, ownership, privilege, membership, resource, same_org):
#     return _get_name('test', **locals())
#
# def is_valid(scope, context, ownership, privilege, membership, resource, same_org):
#     if context == "sandbox" and membership:
#         return False
#     if scope == 'list' and ownership != 'None':
#         return False
#     if context == 'sandbox' and same_org == False:
#         return False
#
#     return True
#
# def gen_test_rego(name):
#     with open(f'{name}_test.gen.rego', 'wt') as f:
#         f.write(f'package {name}\n\n')
#         for scope, context, ownership, privilege, membership, same_org in product(
#             SCOPES, CONTEXTS, OWNERSHIPS, GROUPS, ORG_ROLES, SAME_ORG):
#             for resource in RESOURCES(scope):
#                 if not is_valid(scope, context, ownership, privilege, membership, resource, same_org):
#                     continue
#
#                 data = get_data(scope, context, ownership, privilege, membership, resource, same_org)
#                 test_name = get_name(scope, context, ownership, privilege, membership, resource, same_org)
#                 result = eval_rule(scope, context, ownership, privilege, membership, data)
#                 f.write('{test_name} {{\n    {allow} with input as {data}\n}}\n\n'.format(
#                     test_name=test_name, allow='allow' if result else 'not allow',
#                     data=json.dumps(data)))
#
#         # Write the script which is used to generate the file
#         with open(sys.argv[0]) as this_file:
#             f.write(f'\n\n# {os.path.split(sys.argv[0])[1]}\n')
#             for line in this_file:
#                 if line.strip():
#                     f.write(f'# {line}')
#                 else:
#                     f.write(f'#\n')
#
#         # Write rules which are used to generate the file
#         with open(os.path.join(sys.argv[1], f'{name}.csv')) as rego_file:
#             f.write(f'\n\n# {name}.csv\n')
#             for line in rego_file:
#                 if line.strip():
#                     f.write(f'# {line}')
#                 else:
#                     f.write(f'#\n')
#
# gen_test_rego(NAME)

[k1won]

@nmanovic , I added the jobs.csv and jobs_test.gen.rego files that I modified a while ago.
And I ran opa test on my local dev environment and got the result as below.

$ opa test jobs_test.gen.rego jobs.rego utils.rego organizations.rego
PASS: 3300/3300

[nmanovic]

@k1won , great job! Thanks for all your time you invest into CVAT.