HP-2490: Add login_methods field in ProfileNode

open_city_profile/decorators.py

@@ -8,6 +8,20 @@
 
 
 def _use_context_tests(*test_funcs):
+    """
+    Decorator for running context tests before the decorated function.
+
+    E.g. to create a decorator that checks that the user is authenticated::
+
+        def _require_authenticated(context):
+            # Check that the user is authenticated
+            ...
+
+        @_use_context_tests(_require_authenticated)
+        def login_required():
+            pass
+    """
+
     def decorator(decorator_function):
         @wraps(decorator_function)
         def wrapper(function):
@@ -61,12 +75,12 @@ def permission_checker(context):
 
 
 @_use_context_tests(_require_authenticated)
-def login_required():
+def login_required(*_, **__):
     """Decorator for checking that the user is logged in"""
 
 
 @_use_context_tests(_require_authenticated, _require_service)
-def login_and_service_required():
+def login_and_service_required(*_, **__):
     """Decorator for checking that the user is logged in and service is known"""
 
 

open_city_profile/tests/snapshots/snap_test_graphql_api_schema.py

@@ -122,6 +122,7 @@
   phones(offset: Int, before: String, after: String, first: Int, last: Int): PhoneNodeConnection
   addresses(offset: Int, before: String, after: String, first: Int, last: Int): AddressNodeConnection
   contactMethod: ContactMethod
+  loginMethods: [LoginMethodType]
   sensitivedata: SensitiveDataNode
   serviceConnections(offset: Int, before: String, after: String, first: Int, last: Int): ServiceConnectionTypeConnection
   verifiedPersonalInformation: VerifiedPersonalInformationNode
@@ -215,6 +216,12 @@
   SMS
 }
 
+enum LoginMethodType {
+  PASSWORD
+  OTP
+  SUOMI_FI
+}
+
 type SensitiveDataNode implements Node {
   id: ID!
   ssn: String!

profiles/enums.py

@@ -41,3 +41,14 @@ class Labels:
         WORK = _("Work address")
         HOME = _("Home address")
         OTHER = _("Other address")
+
+
+class LoginMethodType(Enum):
+    PASSWORD = "password"
+    OTP = "otp"
+    SUOMI_FI = "suomi_fi"
+
+    class Labels:
+        PASSWORD = _("Password")
+        OTP = _("One-time password")
+        SUOMI_FI = _("Suomi.fi")

profiles/keycloak_integration.py

@@ -8,7 +8,7 @@
 )
 from utils import keycloak
 
-_keycloak_admin_client = None
+_keycloak_admin_client: keycloak.KeycloakAdminClient | None = None
 
 
 def _setup_keycloak_client():
@@ -103,3 +103,29 @@ def send_profile_changes_to_keycloak(instance):
             _keycloak_admin_client.send_verify_email(user_id)
         except Exception:
             pass
+
+
+def get_user_identity_providers(user_id) -> set[str]:
+    if not _keycloak_admin_client:
+        return set()
+
+    try:
+        user_data = _keycloak_admin_client.get_user_federated_identities(user_id)
+        return {ip["identityProvider"] for ip in user_data}
+    except keycloak.UserNotFoundError:
+        return set()
+
+
+def get_user_credential_types(user_id) -> set[str]:
+    if not _keycloak_admin_client:
+        return set()
+
+    try:
+        user_data = _keycloak_admin_client.get_user_credentials(user_id)
+        return {cred["type"] for cred in user_data}
+    except keycloak.UserNotFoundError:
+        return set()
+
+
+def get_user_login_methods(user_id) -> set[str]:
+    return get_user_identity_providers(user_id) | get_user_credential_types(user_id)

profiles/models.py

@@ -119,6 +119,7 @@ class Meta:
         "verified_personal_information",
         "language",
         "contact_method",
+        "login_methods",
     ]
 
     def resolve_profile(self):

profiles/schema.py

@@ -1,3 +1,4 @@
+import logging
 from itertools import chain
 
 import django.dispatch
@@ -55,8 +56,8 @@
     delete_connected_service_data,
     download_connected_service_data,
 )
-from .enums import AddressType, EmailType, PhoneType
-from .keycloak_integration import delete_profile_from_keycloak
+from .enums import AddressType, EmailType, LoginMethodType, PhoneType
+from .keycloak_integration import delete_profile_from_keycloak, get_user_login_methods
 from .models import (
     Address,
     ClaimToken,
@@ -72,10 +73,14 @@
     VerifiedPersonalInformationTemporaryAddress,
 )
 from .utils import (
+    enum_values,
+    force_list,
     requester_can_view_verified_personal_information,
     requester_has_sufficient_loa_to_perform_gdpr_request,
 )
 
+logger = logging.getLogger(__name__)
+
 User = get_user_model()
 
 AllowedEmailType = graphene.Enum.from_enum(
@@ -514,6 +519,10 @@ def validate_ssn(value, info, **input):
 
 
 class RestrictedProfileNode(DjangoObjectType):
+    """
+    Profile node with a restricted set of data. This does not contain any sensitive data.
+    """
+
     class Meta:
         model = Profile
         fields = ("first_name", "last_name", "nickname", "language")
@@ -575,6 +584,13 @@ class Meta:
         connection_class = ProfilesConnection
         filterset_class = ProfileFilter
 
+    login_methods = graphene.List(
+        graphene.Enum.from_enum(
+            LoginMethodType, description=lambda e: e.label if e else ""
+        ),
+        description="List of login methods that the profile has used to authenticate. "
+        "Only visible to the user themselves.",
+    )
     sensitivedata = graphene.Field(
         SensitiveDataNode,
         description="Data that is consider to be sensitive e.g. social security number",
@@ -589,6 +605,36 @@ class Meta:
         "privileges to access this information.",
     )
 
+    def resolve_login_methods(self: Profile, info, **kwargs):
+        if info.context.user != self.user:
+            raise PermissionDenied(
+                "No permission to read login methods of another user."
+            )
+
+        amr = set(force_list(info.context.user_auth.data.get("amr")))
+
+        # For future software archeologists:
+        # This field was added to the API to support the front-end's need to know
+        # which login methods the user has used. It's only needed for profiles
+        # with helsinki-tunnus or Suomi.fi, so for other cases, save a couple
+        # API calls and return an empty list. There's no other reasoning for the
+        # logic here.
+        # Can remove this after Tunnistamo is no longer in use. Related ticket: HP-2495
+        if amr.intersection({"helsinki_tunnus", "heltunnistussuomifi", "suomi_fi"}):
+            login_methods = get_user_login_methods(self.user.uuid)
+            login_methods_in_enum = {
+                val for val in login_methods if val in enum_values(LoginMethodType)
+            }
+            if unknown_login_methods := login_methods - login_methods_in_enum:
+                logger.warning(
+                    "Found login methods which are not part of the LoginMethodType enum: %s",
+                    unknown_login_methods,
+                )
+
+            return login_methods_in_enum
+
+        return []
+
     def resolve_service_connections(self: Profile, info, **kwargs):
         return self.effective_service_connections_qs()
 

profiles/tests/test_gql_my_profile_query.py

@@ -576,3 +576,91 @@ def test_my_profile_checks_allowed_data_fields_for_multiple_queries(
     assert executed["data"]["myProfile"]["lastName"] == profile.last_name
     assert executed["data"]["myProfile"]["sensitivedata"] is None
     assert executed["data"]["services"] is None
+
+
+@pytest.mark.parametrize(
+    "amr_claim_value", ["suomi_fi", "helsinki_tunnus", "heltunnistussuomifi"]
+)
+def test_user_can_see_own_login_methods_with_correct_amr_claim(
+    user_gql_client, profile, group, service, monkeypatch, amr_claim_value
+):
+    def mock_return(*_, **__):
+        return {"suomi_fi", "password"}
+
+    monkeypatch.setattr(
+        "profiles.keycloak_integration.get_user_identity_providers", mock_return
+    )
+
+    profile = ProfileFactory(user=user_gql_client.user)
+    ServiceConnectionFactory(profile=profile, service=service)
+
+    query = """
+        {
+            myProfile {
+                loginMethods
+            }
+        }
+    """
+    executed = user_gql_client.execute(
+        query, auth_token_payload={"amr": amr_claim_value}, service=service
+    )
+    assert "errors" not in executed
+    assert set(executed["data"]["myProfile"]["loginMethods"]) == {
+        "SUOMI_FI",
+        "PASSWORD",
+    }
+
+
+@pytest.mark.parametrize("amr_claim_value", [None, "helsinkiad"])
+def test_user_cannot_see_own_login_methods_with_other_amr_claims(
+    user_gql_client, profile, group, service, monkeypatch, amr_claim_value
+):
+    def mock_return(*_, **__):
+        return {"this should not show up"}
+
+    monkeypatch.setattr(
+        "profiles.keycloak_integration.get_user_identity_providers", mock_return
+    )
+
+    profile = ProfileFactory(user=user_gql_client.user)
+    ServiceConnectionFactory(profile=profile, service=service)
+
+    query = """
+        {
+            myProfile {
+                loginMethods
+            }
+        }
+    """
+    executed = user_gql_client.execute(
+        query, auth_token_payload={"amr": amr_claim_value}, service=service
+    )
+    assert "errors" not in executed
+    assert executed["data"]["myProfile"]["loginMethods"] == []
+
+
+def test_user_does_not_see_non_enum_login_methods(
+    user_gql_client, profile, group, service, monkeypatch
+):
+    def mock_return(*_, **__):
+        return {"password", "this should not show up"}
+
+    monkeypatch.setattr(
+        "profiles.keycloak_integration.get_user_identity_providers", mock_return
+    )
+
+    profile = ProfileFactory(user=user_gql_client.user)
+    ServiceConnectionFactory(profile=profile, service=service)
+
+    query = """
+        {
+            myProfile {
+                loginMethods
+            }
+        }
+    """
+    executed = user_gql_client.execute(
+        query, auth_token_payload={"amr": "helsinki_tunnus"}, service=service
+    )
+    assert "errors" not in executed
+    assert set(executed["data"]["myProfile"]["loginMethods"]) == {"PASSWORD"}

profiles/tests/test_gql_profiles_query.py

@@ -798,3 +798,29 @@ def test_not_specifying_requesters_service_results_in_permission_denied_error(
     """
     executed = user_gql_client.execute(query)
     assert_match_error_code(executed, "PERMISSION_DENIED_ERROR")
+
+
+def test_staff_user_can_not_query_login_methods_of_other_users(
+    user_gql_client, group, service
+):
+    profile = ProfileFactory()
+    ServiceConnectionFactory(profile=profile, service=service)
+    user = user_gql_client.user
+    user.groups.add(group)
+    assign_perm("can_view_profiles", group, service)
+
+    query = """
+        {
+            profiles {
+                edges {
+                    node {
+                        loginMethods
+                    }
+                }
+            }
+        }
+    """
+
+    executed = user_gql_client.execute(query, service=service)
+    assert "errors" in executed
+    assert_match_error_code(executed, "PERMISSION_DENIED_ERROR")