diff --git a/src/backend/core/tests/conftest.py b/src/backend/core/tests/conftest.py index 8da816a1..6adee998 100644 --- a/src/backend/core/tests/conftest.py +++ b/src/backend/core/tests/conftest.py @@ -1,10 +1,15 @@ """Fixtures for tests in the drive core application""" +import base64 from unittest import mock from django.core.cache import cache import pytest +import responses + +from core import factories +from core.tests.utils.urls import reload_urls USER = "user" TEAM = "team" @@ -25,3 +30,92 @@ def mock_user_teams(): "core.models.User.teams", new_callable=mock.PropertyMock ) as mock_teams: yield mock_teams + + +def resource_server_backend_setup(settings): + """ + A fixture to create a user token for testing. + """ + assert ( + settings.OIDC_RS_BACKEND_CLASS + == "lasuite.oidc_resource_server.backend.ResourceServerBackend" + ) + + settings.OIDC_RESOURCE_SERVER_ENABLED = True + settings.OIDC_RS_CLIENT_ID = "some_client_id" + settings.OIDC_RS_CLIENT_SECRET = "some_client_secret" + + settings.OIDC_OP_URL = "https://oidc.example.com" + settings.OIDC_VERIFY_SSL = False + settings.OIDC_TIMEOUT = 5 + settings.OIDC_PROXY = None + settings.OIDC_OP_JWKS_ENDPOINT = "https://oidc.example.com/jwks" + settings.OIDC_OP_INTROSPECTION_ENDPOINT = "https://oidc.example.com/introspect" + settings.OIDC_RS_SCOPES = ["openid", "groups"] + settings.OIDC_RS_ALLOWED_AUDIENCES = ["some_service_provider"] + + +@pytest.fixture +def resource_server_backend_conf(settings): + """ + A fixture to create a user token for testing. + """ + resource_server_backend_setup(settings) + reload_urls() + + +@pytest.fixture +def resource_server_backend(settings): + """ + A fixture to create a user token for testing. + Including a mocked introspection endpoint. + """ + resource_server_backend_setup(settings) + reload_urls() + + with responses.RequestsMock() as rsps: + rsps.add( + responses.POST, + "https://oidc.example.com/introspect", + json={ + "iss": "https://oidc.example.com", + "aud": "some_client_id", # settings.OIDC_RS_CLIENT_ID + "sub": "very-specific-sub", + "client_id": "some_service_provider", + "scope": "openid groups", + "active": True, + }, + ) + + yield rsps + + +@pytest.fixture +def user_specific_sub(): + """ + A fixture to create a user token for testing. + """ + user = factories.UserFactory(sub="very-specific-sub") + + yield user + + +def build_authorization_bearer(token): + """ + Build an Authorization Bearer header value from a token. + + This can be used like this: + client.post( + ... + HTTP_AUTHORIZATION=f"Bearer {build_authorization_bearer('some_token')}", + ) + """ + return base64.b64encode(token.encode("utf-8")).decode("utf-8") + + +@pytest.fixture +def user_token(): + """ + A fixture to create a user token for testing. + """ + return build_authorization_bearer("some_token") diff --git a/src/backend/core/tests/external_api/items/__init__.py b/src/backend/core/tests/external_api/items/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/backend/core/tests/external_api/items/test_external_api_items.py b/src/backend/core/tests/external_api/items/test_external_api_items.py new file mode 100644 index 00000000..2d94c92b --- /dev/null +++ b/src/backend/core/tests/external_api/items/test_external_api_items.py @@ -0,0 +1,406 @@ +""" +Tests for the Resource Server API for items. + +Not testing external API endpoints that are already tested in the /api +because the resource server viewsets inherit from the api viewsets. + +""" + +from io import BytesIO + +from django.core.files.storage import default_storage + +import pytest +from rest_framework.test import APIClient + +from core import factories, models + +pytestmark = pytest.mark.django_db + +# pylint: disable=unused-argument + + +def test_api_items_retrieve_anonymous_public_standalone(): + """ + Anonymous users should not be allowed to retrieve an item from external + API if resource server is not enabled. + """ + item = factories.ItemFactory(link_reach="public") + + response = APIClient().get(f"/external_api/v1.0/items/{item.id!s}/") + + assert response.status_code == 404 + + +def test_api_items_list_connected_not_resource_server(): + """ + Connected users should not be allowed to list items if resource server is not enabled. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory(item=item, user=user, role="reader") + + response = client.get("/external_api/v1.0/items/") + + assert response.status_code == 404 + + +def test_api_items_list_connected_resource_server( + user_token, resource_server_backend, user_specific_sub +): + """Connected users should be allowed to list items from a resource server.""" + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory(item=item, user=user_specific_sub, role="reader") + + response = client.get("/external_api/v1.0/items/") + + assert response.status_code == 200 + + +def test_api_items_list_connected_resource_server_with_invalid_token( + user_token, resource_server_backend +): + """User with an invalid sub should not be allowed to retrieve items from a resource server.""" + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + response = client.get("/external_api/v1.0/items/") + + assert response.status_code == 401 + + +def test_api_items_retrieve_connected_resource_server_with_wrong_abilities( + user_token, user_specific_sub, resource_server_backend +): + """ + User with wrong abilities should not be allowed to retrieve an item from + a resource server. + """ + + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + + response = client.get(f"/external_api/v1.0/items/{item.id!s}/") + + assert response.status_code == 403 + + +def test_api_items_retrieve_connected_resource_server_using_access_token( + user_token, resource_server_backend, user_specific_sub +): + """ + User with an access token should be allowed to retrieve an item from a resource server. + """ + + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.LinkRoleChoices.READER + ) + + response = client.get(f"/external_api/v1.0/items/{item.id!s}/") + + assert response.status_code == 200 + + +def test_api_items_upload_resource_server_using_access_token( + user_token, resource_server_backend, user_specific_sub +): + """ + User with an access token should be allowed to upload an item to a resource server. + """ + + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + item = factories.ItemFactory( + link_reach=models.LinkReachChoices.RESTRICTED, + type=models.ItemTypeChoices.FOLDER, + ) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.post( + f"/external_api/v1.0/items/{item.id!s}/children/", + { + "type": models.ItemTypeChoices.FILE, + "filename": "file.txt", + }, + ) + + assert response.status_code == 201 + data = response.json() + assert data["type"] == models.ItemTypeChoices.FILE + assert data["filename"] == "file.txt" + assert "policy" in data + child = models.Item.objects.get(id=data["id"]) + + default_storage.save( + child.file_key, + BytesIO(b"my prose"), + ) + + response = client.post(f"/external_api/v1.0/items/{child.id!s}/upload-ended/") + + assert response.status_code == 200 + + +# Non allowed actions on resource server. + + +def test_api_items_delete_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should notbe allowed to delete an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.delete(f"/external_api/v1.0/items/{item.id!s}/") + + assert response.status_code == 403 + + +def test_api_items_hard_delete_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should notbe allowed to hard delete an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.delete(f"/external_api/v1.0/items/{item.id!s}/hard-delete/") + + assert response.status_code == 403 + + +def test_api_items_patch_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should notbe allowed to patch an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.patch( + f"/external_api/v1.0/items/{item.id!s}/", {"title": "new title"} + ) + + assert response.status_code == 403 + + +def test_api_items_put_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should not be allowed to put an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.put( + f"/external_api/v1.0/items/{item.id!s}/", {"title": "new title"} + ) + + assert response.status_code == 403 + + +def test_api_items_move_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should not be allowed to move an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.post( + f"/external_api/v1.0/items/{item.id!s}/move/", + { + "target_item_id": factories.ItemFactory( + link_reach=models.LinkReachChoices.RESTRICTED + ).id + }, + ) + + assert response.status_code == 403 + + +def test_api_items_restore_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should not be allowed to restore an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.post(f"/external_api/v1.0/items/{item.id!s}/restore/") + + assert response.status_code == 403 + + +def test_api_items_trashbin_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should notbe allowed to list the trashbin from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + response = client.get("/external_api/v1.0/items/trashbin/") + + assert response.status_code == 403 + + +def test_api_items_link_configuration_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should not be allowed to update the link configuration + of an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.put( + f"/external_api/v1.0/items/{item.id!s}/link-configuration/", + { + "link_reach": models.LinkReachChoices.RESTRICTED, + "link_role": models.LinkRoleChoices.READER, + }, + ) + + assert response.status_code == 403 + + +def test_api_items_accesses_resource_server_not_allowed( + user_token, resource_server_backend_conf, user_specific_sub +): + """ + Connected users should not be allowed to list the accesses of + an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.get(f"/external_api/v1.0/items/{item.id!s}/accesses/") + + assert response.status_code == 404 + + +def test_api_items_accesses_create_resource_server_not_allowed( + user_token, resource_server_backend_conf, user_specific_sub +): + """ + Connected users should not be allowed to create an access for an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.post( + f"/external_api/v1.0/items/{item.id!s}/accesses/", + {"user_id": factories.UserFactory().id, "role": models.RoleChoices.READER}, + ) + + assert response.status_code == 404 + + +def test_api_items_invitations_resource_server_not_allowed( + user_token, resource_server_backend_conf, user_specific_sub +): + """ + Connected users should not be allowed to list the invitations of an item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.get(f"/external_api/v1.0/items/{item.id!s}/invitations/") + + assert response.status_code == 404 + + +def test_api_items_invitations_create_resource_server_not_allowed( + user_token, resource_server_backend_conf, user_specific_sub +): + """ + Connected users should not be allowed to create an invitation for an + item from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED) + factories.UserItemAccessFactory( + item=item, user=user_specific_sub, role=models.RoleChoices.OWNER + ) + + response = client.post( + f"/external_api/v1.0/items/{item.id!s}/invitations/", + {"email": "test@example.com", "role": models.RoleChoices.READER}, + ) + + assert response.status_code == 404 diff --git a/src/backend/core/tests/external_api/test_external_api_users.py b/src/backend/core/tests/external_api/test_external_api_users.py new file mode 100644 index 00000000..b59dc071 --- /dev/null +++ b/src/backend/core/tests/external_api/test_external_api_users.py @@ -0,0 +1,152 @@ +""" +Tests for the Resource Server API for users. + +Not testing external API endpoints that are already tested in the /api +because the resource server viewsets inherit from the api viewsets. + +""" + +import pytest +from rest_framework.test import APIClient + +from core import factories +from core.api import serializers +from core.tests.utils.urls import reload_urls + +pytestmark = pytest.mark.django_db + +# pylint: disable=unused-argument + + +def test_api_users_me_anonymous_public_standalone(): + """ + Anonymous users should not be allowed to retrieve their own user information from external + API if resource server is not enabled. + """ + reload_urls() + response = APIClient().get("/external_api/v1.0/users/me/") + + assert response.status_code == 404 + + +def test_api_users_me_connected_not_resource_server(): + """ + Connected users should not be allowed to retrieve their own user information from external + API if resource server is not enabled. + """ + reload_urls() + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + response = client.get("/external_api/v1.0/users/me/") + + assert response.status_code == 404 + + +def test_api_users_me_connected_resource_server( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should be allowed to retrieve their own user information from external API + if resource server is enabled. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + response = client.get("/external_api/v1.0/users/me/") + + assert response.status_code == 200 + data = response.json() + assert data["id"] == str(user_specific_sub.id) + assert data["email"] == user_specific_sub.email + + +def test_api_users_me_connected_resource_server_with_invalid_token( + user_token, resource_server_backend +): + """ + Connected users should not be allowed to retrieve their own user information from external API + if resource server is enabled with an invalid token. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + response = client.get("/external_api/v1.0/users/me/") + + assert response.status_code == 401 + + +# Non allowed actions on resource server. + + +def test_api_users_list_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should notbe allowed to list users from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + response = client.get("/external_api/v1.0/users/") + + assert response.status_code == 403 + + +def test_api_users_retrieve_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should notbe allowed to list users from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + other_user = factories.UserFactory() + + response = client.get(f"/external_api/v1.0/users/{other_user.id!s}/") + + assert response.status_code == 403 + + +def test_api_users_put_patch_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should notbe allowed to list users from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + other_user = factories.UserFactory() + + new_user_values = serializers.UserSerializer(instance=factories.UserFactory()).data + response = client.put( + f"/external_api/v1.0/users/{other_user.id!s}/", new_user_values + ) + + assert response.status_code == 403 + + response = client.patch( + f"/external_api/v1.0/users/{other_user.id!s}/", + {"email": "new_email@example.com"}, + ) + + assert response.status_code == 403 + + +def test_api_users_delete_resource_server_not_allowed( + user_token, resource_server_backend, user_specific_sub +): + """ + Connected users should notbe allowed to list users from a resource server. + """ + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}") + + other_user = factories.UserFactory() + + response = client.delete(f"/external_api/v1.0/users/{other_user.id!s}/") + + assert response.status_code == 403 diff --git a/src/backend/core/tests/utils/urls.py b/src/backend/core/tests/utils/urls.py new file mode 100644 index 00000000..ec710f70 --- /dev/null +++ b/src/backend/core/tests/utils/urls.py @@ -0,0 +1,20 @@ +"""Utils for testing URLs.""" + +import importlib + +from django.urls import clear_url_caches + + +def reload_urls(): + """ + Reload the URLs. Since the url are loaded based on a + settings value, we need to reload the urls to make the + URL settings based condition effective. + """ + import core.urls # pylint:disable=import-outside-toplevel # noqa: PLC0415 + + import drive.urls # pylint:disable=import-outside-toplevel # noqa: PLC0415 + + importlib.reload(core.urls) + importlib.reload(drive.urls) + clear_url_caches() diff --git a/src/backend/e2e/tests/test_api_e2e.py b/src/backend/e2e/tests/test_api_e2e.py index ebed3c41..58123a48 100644 --- a/src/backend/e2e/tests/test_api_e2e.py +++ b/src/backend/e2e/tests/test_api_e2e.py @@ -2,26 +2,16 @@ Test e2e API endpoints. """ -import importlib - from django.test.utils import override_settings -from django.urls import clear_url_caches import pytest from rest_framework.test import APIClient +from core.tests.utils.urls import reload_urls + pytestmark = pytest.mark.django_db -def reload_urls(): - """Reload the URLs to test the e2e URLs. Since the url are loaded based on a - settings value, we need to reload the urls to test the e2e urls.""" - import drive.urls # pylint:disable=import-outside-toplevel # noqa: PLC0415 - - importlib.reload(drive.urls) - clear_url_caches() - - def test_api_e2e_user_auth_no_urls(): """E2E URLs not enabled should 404.""" client = APIClient()