feat: prevent GDPR features with insufficient loa

open_city_profile/consts.py

@@ -22,6 +22,7 @@
 SERVICE_NOT_IDENTIFIED_ERROR = "SERVICE_NOT_IDENTIFIED_ERROR"
 PROFILE_ALREADY_EXISTS_FOR_USER_ERROR = "PROFILE_ALREADY_EXISTS_FOR_USER_ERROR"
 PROFILE_MUST_HAVE_PRIMARY_EMAIL = "PROFILE_MUST_HAVE_PRIMARY_EMAIL"
+INSUFFICIENT_LOA_ERROR = "INSUFFICIENT_LOA_ERROR"
 
 # Service specific GDPR errors
 SERVICE_GDPR_API_REQUEST_ERROR = "SERVICE_GDPR_API_REQUEST_ERROR"

open_city_profile/exceptions.py

@@ -69,3 +69,7 @@ class TokenExpiredError(ProfileGraphQLError):
 
 class TokenExchangeError(Exception):
     """OAuth/OIDC token exchange related exception."""
+
+
+class InsufficientLoaError(ProfileGraphQLError):
+    """The requester has insufficient level of authentication to retrieve this data"""

open_city_profile/views.py

@@ -13,6 +13,7 @@
     CONNECTED_SERVICE_DELETION_NOT_ALLOWED_ERROR,
     DATA_CONFLICT_ERROR,
     GENERAL_ERROR,
+    INSUFFICIENT_LOA_ERROR,
     INVALID_EMAIL_FORMAT_ERROR,
     JWT_AUTHENTICATION_ERROR,
     MISSING_GDPR_API_TOKEN_ERROR,
@@ -33,6 +34,7 @@
     ConnectedServiceDeletionFailedError,
     ConnectedServiceDeletionNotAllowedError,
     DataConflictError,
+    InsufficientLoaError,
     InvalidEmailFormatError,
     MissingGDPRApiTokenError,
     ProfileAlreadyExistsForUserError,
@@ -71,6 +73,7 @@
     ServiceAlreadyExistsError: SERVICE_CONNECTION_ALREADY_EXISTS_ERROR,
     ServiceConnectionDoesNotExist: SERVICE_CONNECTION_DOES_NOT_EXIST_ERROR,
     ServiceNotIdentifiedError: SERVICE_NOT_IDENTIFIED_ERROR,
+    InsufficientLoaError: INSUFFICIENT_LOA_ERROR,
 }
 
 sentry_ignored_errors = (

profiles/schema.py

@@ -37,6 +37,7 @@
 from open_city_profile.exceptions import (
     ConnectedServiceDeletionFailedError,
     ConnectedServiceDeletionNotAllowedError,
+    InsufficientLoaError,
     InvalidEmailFormatError,
     ProfileAlreadyExistsForUserError,
     ProfileDoesNotExistError,
@@ -70,7 +71,10 @@
     VerifiedPersonalInformationPermanentForeignAddress,
     VerifiedPersonalInformationTemporaryAddress,
 )
-from .utils import requester_can_view_verified_personal_information
+from .utils import (
+    requester_can_view_verified_personal_information,
+    requester_has_sufficient_loa_to_perform_gdpr_request,
+)
 
 User = get_user_model()
 
@@ -1487,6 +1491,12 @@ def mutate_and_get_payload(cls, root, info, **input):
                 _("You do not have permission to perform this action.")
             )
 
+        if not requester_has_sufficient_loa_to_perform_gdpr_request(info.context):
+            raise InsufficientLoaError(
+                _(
+                    "You have insufficient level of authentication to perform this action."
+                )
+            )
         dry_run = input.get("dry_run", False)
 
         results = delete_connected_service_data(
@@ -1551,6 +1561,13 @@ def mutate(parent, info, input):
                 _("You do not have permission to perform this action.")
             )
 
+        if not requester_has_sufficient_loa_to_perform_gdpr_request(info.context):
+            raise InsufficientLoaError(
+                _(
+                    "You have insufficient level of authentication to perform this action."
+                )
+            )
+
         service_connections = profile.effective_service_connections_qs().filter(
             service__name=input["service_name"]
         )
@@ -1724,6 +1741,13 @@ def resolve_download_my_profile(self, info, **kwargs):
                 _("You do not have permission to perform this action.")
             )
 
+        if not requester_has_sufficient_loa_to_perform_gdpr_request(info.context):
+            raise InsufficientLoaError(
+                _(
+                    "You have insufficient level of authentication to perform this action."
+                )
+            )
+
         external_data = download_connected_service_data(
             profile,
             kwargs["authorization_code"],

profiles/tests/gdpr/test_gql_delete_my_profile_mutation.py

@@ -11,7 +11,11 @@
 from open_city_profile.oidc import TunnistamoTokenExchange
 from open_city_profile.tests.asserts import assert_match_error_code
 from profiles.models import Profile
-from profiles.tests.factories import ProfileFactory, ProfileWithPrimaryEmailFactory
+from profiles.tests.factories import (
+    ProfileFactory,
+    ProfileWithPrimaryEmailFactory,
+    VerifiedPersonalInformationFactory,
+)
 from services.models import ServiceConnection
 from services.tests.factories import ServiceConnectionFactory
 from users.models import User
@@ -546,3 +550,26 @@ def test_api_tokens_missing(user_gql_client, service_1, mocker):
     executed = user_gql_client.execute(DELETE_MY_PROFILE_MUTATION)
 
     assert_match_error_code(executed, MISSING_GDPR_API_TOKEN_ERROR)
+
+
+@pytest.mark.parametrize(
+    "loa,errors",
+    [
+        ("low", True),
+        ("substantial", False),
+        ("high", False),
+    ],
+)
+def test_delete_with_verified_personal_information(
+    user_gql_client, gdpr_api_tokens, loa, errors
+):
+    """Deletion is allowed when GDPR URL is set, and service returns a successful status."""
+    profile = ProfileFactory(user=user_gql_client.user)
+    VerifiedPersonalInformationFactory(profile=profile)
+
+    executed = user_gql_client.execute(
+        DELETE_MY_PROFILE_MUTATION, auth_token_payload={"loa": loa}
+    )
+    assert bool(executed.get("errors")) == errors
+    if errors:
+        assert executed["errors"][0]["extensions"]["code"] == "INSUFFICIENT_LOA_ERROR"

profiles/tests/gdpr/test_gql_download_my_profile_query.py

@@ -2,7 +2,6 @@
 from string import Template
 
 import pytest
-from django.utils.translation import gettext as _
 
 from open_city_profile.consts import MISSING_GDPR_API_TOKEN_ERROR
 from open_city_profile.oidc import TunnistamoTokenExchange
@@ -128,6 +127,7 @@ def download_verified_personal_information_with_loa(loa, user_gql_client, servic
     )
 
     full_dump = json.loads(executed["data"]["downloadMyProfile"])
+
     profile_dump = next(
         child for child in full_dump["children"] if child["key"] == "PROFILE"
     )
@@ -154,16 +154,20 @@ def test_verified_personal_information_is_included_in_the_downloaded_profile_whe
 
 @pytest.mark.parametrize("loa", [None, "foo", "low"])
 def test_verified_personal_information_is_replaced_with_an_error_when_loa_is_not_high_enough(
-    loa, user_gql_client, profile_service
+    loa, user_gql_client, profile_service, service
 ):
-    vpi_dump = download_verified_personal_information_with_loa(
-        loa, user_gql_client, profile_service
-    )
-
-    assert vpi_dump == {
-        "key": "VERIFIEDPERSONALINFORMATION",
-        "error": _("No permission to read verified personal information."),
+    profile = ProfileFactory(user=user_gql_client.user)
+    ServiceConnectionFactory(profile=profile, service=service)
+    VerifiedPersonalInformationFactory(profile=profile)
+    token_payload = {
+        "loa": loa,
     }
+    executed = user_gql_client.execute(
+        DOWNLOAD_MY_PROFILE_MUTATION, service=service, auth_token_payload=token_payload
+    )
+    assert executed["data"]["downloadMyProfile"] is None
+    assert len(executed["errors"]) == 1
+    assert executed["errors"][0]["extensions"]["code"] == "INSUFFICIENT_LOA_ERROR"
 
 
 def test_downloading_non_existent_profile_doesnt_return_errors(user_gql_client):

profiles/tests/test_audit_log_to_db.py

@@ -414,7 +414,9 @@ def test_audit_log_delete(live_server, profile_with_related):
             }
         }
     """
-    do_graphql_call_as_user(live_server, user, query=query)
+    do_graphql_call_as_user(
+        live_server, user, query=query, extra_claims={"loa": "substantial"}
+    )
 
     log_entries = list(LogEntry.objects.all())
     # Audit logging the Profile DELETE with a related object causes some READs

profiles/tests/test_audit_log_to_logger.py

@@ -442,7 +442,9 @@ def test_audit_log_delete(live_server, profile_with_related, cap_audit_log):
             }
         }
     """
-    do_graphql_call_as_user(live_server, user, query=query)
+    do_graphql_call_as_user(
+        live_server, user, query=query, extra_claims={"loa": "substantial"}
+    )
 
     audit_logs = cap_audit_log.get_logs()
     # Audit logging the Profile DELETE with a related object causes some READs

profiles/utils.py

@@ -29,3 +29,15 @@ def requester_can_view_verified_personal_information(request):
         or request.user_auth.data.get("amr")
         in settings.VERIFIED_PERSONAL_INFORMATION_ACCESS_AMR_LIST
     )
+
+
+def requester_has_sufficient_loa_to_perform_gdpr_request(request):
+    user = request.user
+    profile = user.profile
+    if not profile:
+        return False
+    loa = request.user_auth.data.get("loa")
+    return not hasattr(profile, "verified_personal_information") or loa in [
+        "substantial",
+        "high",
+    ]