wehub-resource-sync
481 lines
20 KiB
Python
481 lines
20 KiB
Python
import csv
|
|
import json
|
|
from contextlib import ExitStack
|
|
from datetime import datetime, timedelta, timezone
|
|
from http import HTTPStatus
|
|
from io import StringIO
|
|
from time import sleep
|
|
|
|
import pytest
|
|
from cvat_sdk.api_client import exceptions, models
|
|
from cvat_sdk.api_client.api_client import ApiClient, Endpoint
|
|
from deepdiff import DeepDiff
|
|
from pytest_cases import parametrize
|
|
|
|
from shared.utils.config import make_api_client
|
|
|
|
from .utils import CollectionSimpleFilterTestBase, export_backup, export_dataset, export_events
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_function")
|
|
class TestPostAccessToken:
|
|
@pytest.mark.parametrize("user_group", ["admin", "user", "worker"])
|
|
def test_can_create_token(self, users, user_group):
|
|
user = next(u for u in users if user_group in u["groups"])
|
|
|
|
with make_api_client(user["username"]) as api_client:
|
|
token, _ = api_client.auth_api.create_access_tokens(
|
|
access_token_write_request=models.AccessTokenWriteRequest(name="test token")
|
|
)
|
|
|
|
assert token.value
|
|
|
|
with make_api_client(access_token=token.value) as api_client:
|
|
user, _ = api_client.users_api.retrieve_self()
|
|
|
|
assert user.username == user["username"]
|
|
|
|
def test_can_create_token_with_expiration(self, admin_user):
|
|
expiration_date = datetime.now(tz=timezone.utc) - timedelta(seconds=1)
|
|
|
|
with make_api_client(admin_user) as api_client:
|
|
token, _ = api_client.auth_api.create_access_tokens(
|
|
access_token_write_request=models.AccessTokenWriteRequest(
|
|
name="test token", expiry_date=expiration_date
|
|
)
|
|
)
|
|
|
|
assert token.expiry_date == expiration_date
|
|
|
|
with (
|
|
make_api_client(access_token=token.value) as api_client,
|
|
pytest.raises(exceptions.UnauthorizedException, match="Invalid token"),
|
|
):
|
|
api_client.users_api.retrieve_self()
|
|
|
|
@parametrize("is_readonly", [True, False])
|
|
def test_can_create_readonly_token(self, admin_user, is_readonly, tasks):
|
|
with make_api_client(admin_user) as api_client:
|
|
token, _ = api_client.auth_api.create_access_tokens(
|
|
access_token_write_request=models.AccessTokenWriteRequest(
|
|
name="test token", read_only=is_readonly
|
|
)
|
|
)
|
|
|
|
assert token.read_only == is_readonly
|
|
|
|
with make_api_client(access_token=token.value) as api_client:
|
|
user, _ = api_client.users_api.retrieve_self()
|
|
assert user.username == admin_user
|
|
|
|
with ExitStack() as es:
|
|
if is_readonly:
|
|
es.enter_context(pytest.raises(exceptions.ForbiddenException))
|
|
|
|
api_client.tasks_api.partial_update(
|
|
next(iter(tasks))["id"],
|
|
patched_task_write_request=models.PatchedTaskWriteRequest(name="new name"),
|
|
)
|
|
|
|
|
|
class TestGetAccessToken:
|
|
def test_can_get_access_token(self, admin_user, access_tokens):
|
|
expected = next(iter(access_tokens))
|
|
|
|
with make_api_client(admin_user) as api_client:
|
|
_, response = api_client.auth_api.retrieve_access_tokens(expected["id"])
|
|
actual = json.loads(response.data)
|
|
|
|
assert DeepDiff(expected, actual, exclude_paths=["private_key"]) == {}
|
|
|
|
@parametrize("is_admin", [True, False])
|
|
def test_cannot_see_foreign_tokens(self, users, access_tokens_by_username, is_admin):
|
|
token_owner, token_owner_tokens = next(iter(access_tokens_by_username.items()))
|
|
token = token_owner_tokens[0]
|
|
|
|
other_user = next(
|
|
u
|
|
for u in users
|
|
if u["username"] != token_owner
|
|
if u["is_superuser"] == is_admin
|
|
if not access_tokens_by_username.get(u["username"])
|
|
)
|
|
|
|
with make_api_client(other_user["username"]) as api_client:
|
|
_, response = api_client.auth_api.retrieve_access_tokens(
|
|
token["id"], _check_status=False
|
|
)
|
|
if is_admin:
|
|
assert response.status == HTTPStatus.OK
|
|
else:
|
|
assert response.status == HTTPStatus.FORBIDDEN
|
|
|
|
assert api_client.auth_api.list_access_tokens()[0].count == 0
|
|
|
|
def test_can_get_self(self, access_tokens_by_username):
|
|
_, user_tokens = next(iter(access_tokens_by_username.items()))
|
|
token = user_tokens[0]
|
|
|
|
with make_api_client(access_token=token["private_key"]) as api_client:
|
|
received_token = json.loads(api_client.auth_api.retrieve_access_tokens_self()[1].data)
|
|
|
|
assert (
|
|
DeepDiff(
|
|
token,
|
|
received_token,
|
|
exclude_paths=["updated_date", "last_used_date", "private_key"],
|
|
)
|
|
== {}
|
|
)
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_function")
|
|
@pytest.mark.parametrize("token_eol_reason", ["expired", "stale", "revoked"])
|
|
def test_can_only_see_alive_tokens(self, token_eol_reason: str, admin_user):
|
|
with make_api_client(admin_user) as api_client:
|
|
if token_eol_reason == "expired":
|
|
token_id = api_client.auth_api.create_access_tokens(
|
|
access_token_write_request=models.AccessTokenWriteRequest(
|
|
name="test token",
|
|
expiry_date=datetime.now(timezone.utc) - timedelta(seconds=1),
|
|
)
|
|
)[0].id
|
|
elif token_eol_reason == "revoked":
|
|
token_id = api_client.auth_api.create_access_tokens(
|
|
access_token_write_request=models.AccessTokenWriteRequest(name="test token")
|
|
)[0].id
|
|
api_client.auth_api.destroy_access_tokens(token_id)
|
|
elif token_eol_reason == "stale":
|
|
token_id = 6
|
|
else:
|
|
assert False, f"Unexpected token eol reason '{token_eol_reason}'"
|
|
|
|
with pytest.raises(
|
|
exceptions.NotFoundException, match="No AccessToken matches the given query"
|
|
):
|
|
api_client.auth_api.retrieve_access_tokens(token_id)
|
|
|
|
assert (
|
|
api_client.auth_api.list_access_tokens(
|
|
filter=json.dumps({"==": [{"var": "id"}, token_id]})
|
|
)[0].count
|
|
== 0
|
|
)
|
|
|
|
|
|
class TestAccessTokenListFilters(CollectionSimpleFilterTestBase):
|
|
@pytest.fixture(autouse=True)
|
|
def setup(self, restore_db_per_class, users_by_name, raw_access_tokens_by_username):
|
|
# Only own keys are visible to each user
|
|
self.user, self.samples = next(
|
|
(username, user_tokens)
|
|
for username, user_tokens in raw_access_tokens_by_username.items()
|
|
if users_by_name[username]["is_superuser"] and user_tokens
|
|
)
|
|
|
|
def _get_endpoint(self, api_client: ApiClient) -> Endpoint:
|
|
return api_client.auth_api.list_access_tokens_endpoint
|
|
|
|
@pytest.mark.parametrize("field", ("name", "read_only"))
|
|
def test_can_use_simple_filter_for_object_list(self, field):
|
|
return super()._test_can_use_simple_filter_for_object_list(field)
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_function")
|
|
class TestPatchAccessToken:
|
|
def test_can_modify_token(self, access_tokens_by_username):
|
|
user, user_tokens = next(iter(access_tokens_by_username.items()))
|
|
token = user_tokens[0]
|
|
|
|
updated_values = {
|
|
"name": "new name",
|
|
"expiry_date": datetime.now(timezone.utc) + timedelta(days=1),
|
|
"read_only": not token["read_only"],
|
|
}
|
|
|
|
updated_token = dict(token)
|
|
updated_token.update(updated_values)
|
|
|
|
with make_api_client(user) as api_client:
|
|
_, response = api_client.auth_api.partial_update_access_tokens(
|
|
token["id"],
|
|
patched_access_token_write_request=models.PatchedAccessTokenWriteRequest(
|
|
**updated_values
|
|
),
|
|
)
|
|
received_token = json.loads(response.data)
|
|
|
|
updated_token["expiry_date"] = (
|
|
updated_token["expiry_date"].isoformat().replace("+00:00", "Z")
|
|
)
|
|
assert (
|
|
DeepDiff(updated_token, received_token, exclude_paths=["updated_date", "private_key"])
|
|
== {}
|
|
)
|
|
|
|
def test_cannot_modify_foreign_token(self, users, access_tokens_by_username):
|
|
token_owner, token_owner_tokens = next(iter(access_tokens_by_username.items()))
|
|
token = token_owner_tokens[0]
|
|
|
|
other_user = next(
|
|
u["username"] for u in users if u["username"] != token_owner and not u["is_superuser"]
|
|
)
|
|
|
|
with (
|
|
make_api_client(other_user) as api_client,
|
|
pytest.raises(exceptions.ForbiddenException),
|
|
):
|
|
api_client.auth_api.partial_update_access_tokens(
|
|
token["id"],
|
|
patched_access_token_write_request=models.PatchedAccessTokenWriteRequest(
|
|
name="new name"
|
|
),
|
|
)
|
|
|
|
def test_cannot_modify_expired_token(self, admin_user):
|
|
with make_api_client(admin_user) as api_client:
|
|
token_id = api_client.auth_api.create_access_tokens(
|
|
access_token_write_request=models.AccessTokenWriteRequest(
|
|
name="test token",
|
|
expiry_date=datetime.now(timezone.utc) - timedelta(seconds=1),
|
|
)
|
|
)[0].id
|
|
|
|
with pytest.raises(
|
|
exceptions.NotFoundException, match="No AccessToken matches the given query"
|
|
):
|
|
api_client.auth_api.partial_update_access_tokens(
|
|
token_id,
|
|
patched_access_token_write_request=models.PatchedAccessTokenWriteRequest(
|
|
expiry_date=datetime.now(timezone.utc) + timedelta(days=1)
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_function")
|
|
class TestDeleteAccessToken:
|
|
def test_can_revoke_own_token(self, access_tokens_by_username):
|
|
user, user_tokens = next(iter(access_tokens_by_username.items()))
|
|
token = user_tokens[0]
|
|
|
|
with make_api_client(user) as api_client:
|
|
api_client.auth_api.destroy_access_tokens(token["id"])
|
|
|
|
def test_cannot_revoke_foreign_token(self, users, access_tokens_by_username):
|
|
token_owner, token_owner_tokens = next(iter(access_tokens_by_username.items()))
|
|
token = token_owner_tokens[0]
|
|
|
|
other_user = next(
|
|
u["username"] for u in users if u["username"] != token_owner and not u["is_superuser"]
|
|
)
|
|
|
|
with (
|
|
make_api_client(other_user) as api_client,
|
|
pytest.raises(exceptions.ForbiddenException),
|
|
):
|
|
api_client.auth_api.destroy_access_tokens(token["id"])
|
|
|
|
|
|
class TestTokenAuthPermissions:
|
|
# Some operations are not allowed with token auth for security reasons, even if not read only
|
|
|
|
def test_cannot_change_password_when_using_token_auth(
|
|
self, admin_user, access_tokens_by_username
|
|
):
|
|
token = next(t for t in access_tokens_by_username[admin_user] if not t["read_only"])
|
|
|
|
with (
|
|
make_api_client(access_token=token["private_key"]) as api_client,
|
|
pytest.raises(exceptions.UnauthorizedException),
|
|
):
|
|
api_client.auth_api.create_password_change(
|
|
password_change_request=models.PasswordChangeRequest(
|
|
old_password="any", new_password1="anypass1", new_password2="anypass1"
|
|
)
|
|
)
|
|
|
|
def test_cannot_create_token_when_using_token_auth(self, admin_user, access_tokens_by_username):
|
|
token = next(t for t in access_tokens_by_username[admin_user] if not t["read_only"])
|
|
|
|
with (
|
|
make_api_client(access_token=token["private_key"]) as api_client,
|
|
pytest.raises(exceptions.ForbiddenException),
|
|
):
|
|
api_client.auth_api.create_access_tokens(
|
|
access_token_write_request=models.AccessTokenWriteRequest(name="test token")
|
|
)
|
|
|
|
def test_cannot_edit_token_when_using_token_auth(self, admin_user, access_tokens_by_username):
|
|
token = next(t for t in access_tokens_by_username[admin_user] if not t["read_only"])
|
|
|
|
with (
|
|
make_api_client(access_token=token["private_key"]) as api_client,
|
|
pytest.raises(exceptions.ForbiddenException),
|
|
):
|
|
api_client.auth_api.partial_update_access_tokens(
|
|
token["id"],
|
|
patched_access_token_write_request=models.PatchedAccessTokenWriteRequest(
|
|
name="new name"
|
|
),
|
|
)
|
|
|
|
def test_cannot_revoke_token_when_using_token_auth(self, admin_user, access_tokens_by_username):
|
|
token = next(t for t in access_tokens_by_username[admin_user] if not t["read_only"])
|
|
|
|
with (
|
|
make_api_client(access_token=token["private_key"]) as api_client,
|
|
pytest.raises(exceptions.ForbiddenException),
|
|
):
|
|
api_client.auth_api.destroy_access_tokens(token["id"])
|
|
|
|
def test_cannot_edit_user_when_using_token_auth(
|
|
self, admin_user, access_tokens_by_username, users_by_name
|
|
):
|
|
token = next(t for t in access_tokens_by_username[admin_user] if not t["read_only"])
|
|
user = users_by_name[admin_user]
|
|
|
|
with (
|
|
make_api_client(access_token=token["private_key"]) as api_client,
|
|
pytest.raises(exceptions.ForbiddenException),
|
|
):
|
|
api_client.users_api.partial_update(
|
|
user["id"], patched_user_request=models.PatchedUserRequest(first_name="new name")
|
|
)
|
|
|
|
@pytest.mark.usefixtures("restore_redis_ondisk_per_function")
|
|
@parametrize("is_readonly", [True, False])
|
|
@parametrize("export_type", ["annotations", "dataset", "backup"])
|
|
def test_token_can_export_project(
|
|
self, admin_user, access_tokens_by_username, projects, export_type, is_readonly
|
|
):
|
|
token = next(
|
|
t for t in access_tokens_by_username[admin_user] if t["read_only"] == is_readonly
|
|
)
|
|
|
|
project_id = next(p["id"] for p in projects if p["tasks"]["count"] > 0)
|
|
|
|
with make_api_client(access_token=token["private_key"]) as api_client:
|
|
if export_type in ("annotations", "dataset"):
|
|
export_dataset(
|
|
api_client.projects_api, id=project_id, save_images=(export_type == "dataset")
|
|
)
|
|
elif export_type == "backup":
|
|
export_backup(api_client.projects_api, id=project_id)
|
|
|
|
@pytest.mark.usefixtures("restore_redis_ondisk_per_function")
|
|
@parametrize("is_readonly", [True, False])
|
|
@parametrize("export_type", ["annotations", "dataset", "backup"])
|
|
def test_token_can_export_task(
|
|
self, admin_user, access_tokens_by_username, tasks, export_type, is_readonly
|
|
):
|
|
token = next(
|
|
t for t in access_tokens_by_username[admin_user] if t["read_only"] == is_readonly
|
|
)
|
|
|
|
task_id = next(p["id"] for p in tasks if p["size"] > 0 if p["media_type"] != "audio")
|
|
|
|
with make_api_client(access_token=token["private_key"]) as api_client:
|
|
if export_type in ("annotations", "dataset"):
|
|
export_dataset(
|
|
api_client.tasks_api, id=task_id, save_images=(export_type == "dataset")
|
|
)
|
|
elif export_type == "backup":
|
|
export_backup(api_client.tasks_api, id=task_id)
|
|
|
|
@pytest.mark.usefixtures("restore_redis_ondisk_per_function")
|
|
@parametrize("is_readonly", [True, False])
|
|
@parametrize("export_type", ["annotations", "dataset"])
|
|
def test_token_can_export_jobs(
|
|
self, admin_user, access_tokens_by_username, jobs, export_type, is_readonly
|
|
):
|
|
token = next(
|
|
t for t in access_tokens_by_username[admin_user] if t["read_only"] == is_readonly
|
|
)
|
|
|
|
job_id = next(p["id"] for p in jobs if p["media_type"] != "audio")
|
|
|
|
with make_api_client(access_token=token["private_key"]) as api_client:
|
|
if export_type in ("annotations", "dataset"):
|
|
export_dataset(
|
|
api_client.jobs_api, id=job_id, save_images=(export_type == "dataset")
|
|
)
|
|
elif export_type == "backup":
|
|
export_backup(api_client.jobs_api, id=job_id)
|
|
|
|
|
|
@pytest.mark.usefixtures("restore_db_per_function")
|
|
class TestTokenTracking:
|
|
def test_can_update_last_use(self, access_tokens_by_username):
|
|
_, user_tokens = next(iter(access_tokens_by_username.items()))
|
|
token = user_tokens[0]
|
|
|
|
with make_api_client(access_token=token["private_key"]) as api_client:
|
|
updated_token, _ = api_client.auth_api.retrieve_access_tokens_self()
|
|
|
|
old_last_used_date = token["last_used_date"]
|
|
if old_last_used_date is not None:
|
|
old_last_used_date = datetime.fromisoformat(
|
|
token["last_used_date"].rstrip("Z")
|
|
).replace(tzinfo=timezone.utc)
|
|
|
|
assert updated_token.last_used_date != old_last_used_date
|
|
|
|
@pytest.mark.usefixtures("restore_redis_inmem_per_function")
|
|
def test_can_record_token_events_in_audit_logs(self, admin_user, tasks):
|
|
test_start_date = datetime.now(tz=timezone.utc)
|
|
|
|
with make_api_client(admin_user) as api_client:
|
|
token = api_client.auth_api.create_access_tokens(
|
|
access_token_write_request=models.AccessTokenWriteRequest(
|
|
name="test token", read_only=False
|
|
)
|
|
)[0]
|
|
|
|
task_id = next(p["id"] for p in tasks if p["size"] > 0)
|
|
|
|
with make_api_client(access_token=token.value) as api_client:
|
|
api_client.tasks_api.partial_update(
|
|
task_id, patched_task_write_request=models.PatchedTaskWriteRequest(name="newname")
|
|
)
|
|
|
|
with make_api_client(admin_user) as api_client:
|
|
api_client.auth_api.partial_update_access_tokens(
|
|
token.id,
|
|
patched_access_token_write_request=models.PatchedAccessTokenWriteRequest(
|
|
expiry_date=test_start_date + timedelta(days=1)
|
|
),
|
|
)
|
|
|
|
api_client.auth_api.destroy_access_tokens(token.id)
|
|
|
|
# Clickhouse updates are not immediate, vector has buffering on sinks.
|
|
# All these checks are in a single test because of this extra waiting.
|
|
# Potentially unstable, should probably be improved or maybe moved into server tests.
|
|
sleep(5)
|
|
|
|
events_csv = export_events(api_client, api_version=2, _from=test_start_date)
|
|
csv_reader = csv.DictReader(StringIO(events_csv.decode()))
|
|
rows = list(csv_reader)
|
|
|
|
def find_row(scope: str):
|
|
return next((i, r) for i, r in enumerate(rows) if r["scope"] == scope)
|
|
|
|
token_creation_row_index, row = find_row(scope="create:accesstoken")
|
|
assert row["obj_id"] == str(token.id)
|
|
|
|
task_update_row_index, row = find_row(scope="update:task")
|
|
assert row["task_id"] == str(task_id)
|
|
assert row["access_token_id"] == str(token.id)
|
|
assert row["obj_name"] == "name"
|
|
assert row["obj_val"] == "newname"
|
|
|
|
token_update_row_index, row = find_row(scope="update:accesstoken")
|
|
# token id is not present in the "update:*" event fields, can't check it
|
|
assert row["obj_name"] == "expiry_date"
|
|
|
|
token_delete_row_index, row = find_row(scope="delete:accesstoken")
|
|
assert row["obj_id"] == str(token.id)
|
|
|
|
assert token_creation_row_index < task_update_row_index
|
|
assert task_update_row_index < token_update_row_index
|
|
assert token_update_row_index < token_delete_row_index
|