service.py

import dataclasses
import logging
import time
import uuid
from collections import ChainMap
from collections.abc import Mapping
from datetime import datetime
from typing import Any, cast

from aiohttp import web
from sqlalchemy import RowMapping

from ai.backend.common.clients.valkey_client.valkey_session.client import ValkeySessionClient
from ai.backend.common.clients.valkey_client.valkey_session.types import (
    LoginSessionData,
    LoginSessionInner,
    LoginSessionTokenData,
)
from ai.backend.common.dto.manager.auth.types import AuthTokenType
from ai.backend.common.exception import InvalidAPIParameters, UserResourcePolicyNotFound
from ai.backend.common.identifier.user import UserID
from ai.backend.common.plugin.hook import ALL_COMPLETED, FIRST_COMPLETED, PASSED, HookPluginContext
from ai.backend.common.types import AccessKey, SecretKey, SSHPrivateKey, SSHPublicKey
from ai.backend.logging.utils import BraceStyleAdapter
from ai.backend.manager.config.provider import ManagerConfigProvider
from ai.backend.manager.config.unified import AuthConfig
from ai.backend.manager.data.auth.login_session_types import LoginAttemptResult
from ai.backend.manager.data.auth.types import AuthorizationResult, SSHKeypair
from ai.backend.manager.defs import DEFAULT_PROJECT_NAME
from ai.backend.manager.errors.auth import (
    AuthorizationFailed,
    EmailAlreadyExistsError,
    GroupMembershipNotFoundError,
    PasswordExpired,
    TooManyConcurrentLoginSessions,
)
from ai.backend.manager.errors.common import (
    GenericBadRequest,
    GenericForbidden,
    ObjectNotFound,
    RejectedByHook,
)
from ai.backend.manager.models.hasher.types import PasswordInfo
from ai.backend.manager.models.keypair import (
    generate_ssh_keypair,
)
from ai.backend.manager.models.keypair.ssh_key_validator import SSHKeyValidator
from ai.backend.manager.models.user import (
    INACTIVE_USER_STATUSES,
    UserRole,
    UserStatus,
    compare_to_hashed_password,
)
from ai.backend.manager.repositories.auth.db_source.db_source import ActiveSessionInfo
from ai.backend.manager.repositories.auth.repository import AuthRepository
from ai.backend.manager.repositories.group.repository import GroupRepository
from ai.backend.manager.repositories.user.creators import UserCreatorSpec
from ai.backend.manager.repositories.user.repository import UserRepository
from ai.backend.manager.repositories.user_resource_policy.repository import (
    UserResourcePolicyRepository,
)
from ai.backend.manager.services.auth.actions.authorize import (
    AuthorizeAction,
    AuthorizeActionResult,
)
from ai.backend.manager.services.auth.actions.generate_ssh_keypair import (
    GenerateSSHKeypairAction,
    GenerateSSHKeypairActionResult,
)
from ai.backend.manager.services.auth.actions.get_role import GetRoleAction, GetRoleActionResult
from ai.backend.manager.services.auth.actions.get_ssh_keypair import (
    GetSSHKeypairAction,
    GetSSHKeypairActionResult,
)
from ai.backend.manager.services.auth.actions.logout import LogoutAction, LogoutActionResult
from ai.backend.manager.services.auth.actions.resolve_access_key_scope import (
    ResolveAccessKeyScopeAction,
    ResolveAccessKeyScopeResult,
)
from ai.backend.manager.services.auth.actions.resolve_user_id_by_access_key import (
    ResolveUserIDByAccessKeyAction,
    ResolveUserIDByAccessKeyResult,
)
from ai.backend.manager.services.auth.actions.resolve_user_scope import (
    ResolveUserScopeAction,
    ResolveUserScopeResult,
)
from ai.backend.manager.services.auth.actions.revoke_login_session import (
    AdminRevokeLoginSessionAction,
    MyRevokeLoginSessionAction,
    RevokeLoginSessionActionResult,
)
from ai.backend.manager.services.auth.actions.search_login_history import (
    AdminSearchLoginHistoryAction,
    SearchLoginHistoryAction,
    SearchLoginHistoryActionResult,
)
from ai.backend.manager.services.auth.actions.search_login_sessions import (
    AdminSearchLoginSessionsAction,
    SearchLoginSessionsAction,
    SearchLoginSessionsActionResult,
)
from ai.backend.manager.services.auth.actions.signout import SignoutAction, SignoutActionResult
from ai.backend.manager.services.auth.actions.signup import SignupAction, SignupActionResult
from ai.backend.manager.services.auth.actions.unblock_user import (
    AdminUnblockUserAction,
    AdminUnblockUserActionResult,
)
from ai.backend.manager.services.auth.actions.update_full_name import (
    UpdateFullNameAction,
    UpdateFullNameActionResult,
)
from ai.backend.manager.services.auth.actions.update_password import (
    UpdatePasswordAction,
    UpdatePasswordActionResult,
)
from ai.backend.manager.services.auth.actions.update_password_no_auth import (
    UpdatePasswordNoAuthAction,
    UpdatePasswordNoAuthActionResult,
)
from ai.backend.manager.services.auth.actions.upload_ssh_keypair import (
    UploadSSHKeypairAction,
    UploadSSHKeypairActionResult,
)
from ai.backend.manager.utils import check_if_requester_is_eligible_to_act_as_target_user

log = BraceStyleAdapter(logging.getLogger(__spec__.name))

_FAILURE_MAP: dict[type[Exception], LoginAttemptResult] = {
    AuthorizationFailed: LoginAttemptResult.FAILED_INVALID_CREDENTIALS,
    PasswordExpired: LoginAttemptResult.FAILED_PASSWORD_EXPIRED,
    RejectedByHook: LoginAttemptResult.FAILED_REJECTED_BY_HOOK,
    TooManyConcurrentLoginSessions: LoginAttemptResult.FAILED_SESSION_ALREADY_EXISTS,
}


def _classify_failure(exc: Exception) -> LoginAttemptResult:
    return _FAILURE_MAP.get(type(exc), LoginAttemptResult.FAILED_INVALID_CREDENTIALS)


class AuthService:
    _hook_plugin_ctx: HookPluginContext
    _auth_repository: AuthRepository
    _config_provider: ManagerConfigProvider
    _valkey_session_client: ValkeySessionClient
    _user_resource_policy_repository: UserResourcePolicyRepository
    _user_repository: UserRepository
    _group_repository: GroupRepository
    _ssh_key_validator: SSHKeyValidator

    def __init__(
        self,
        hook_plugin_ctx: HookPluginContext,
        auth_repository: AuthRepository,
        config_provider: ManagerConfigProvider,
        valkey_session_client: ValkeySessionClient,
        user_resource_policy_repository: UserResourcePolicyRepository,
        user_repository: UserRepository,
        group_repository: GroupRepository,
        ssh_key_validator: SSHKeyValidator,
    ) -> None:
        self._hook_plugin_ctx = hook_plugin_ctx
        self._auth_repository = auth_repository
        self._config_provider = config_provider
        self._valkey_session_client = valkey_session_client
        self._user_resource_policy_repository = user_resource_policy_repository
        self._user_repository = user_repository
        self._group_repository = group_repository
        self._ssh_key_validator = ssh_key_validator

    async def get_role(self, action: GetRoleAction) -> GetRoleActionResult:
        group_role = None
        if action.group_id is not None:
            if action.is_superadmin:
                # Superadmins have global access across all domains and groups.
                group_role = "user"
            else:
                try:
                    # TODO: per-group role is not yet implemented.
                    await self._auth_repository.get_group_membership(
                        action.group_id, action.user_id
                    )
                    group_role = "user"
                except GroupMembershipNotFoundError as e:
                    raise ObjectNotFound(
                        extra_msg="No such project or you are not the member of it.",
                        object_name="project (user group)",
                    ) from e

        return GetRoleActionResult(
            global_role="superadmin" if action.is_superadmin else "user",
            domain_role="admin" if action.is_admin else "user",
            group_role=group_role,
        )

    async def authorize(self, action: AuthorizeAction) -> AuthorizeActionResult:
        if action.type != AuthTokenType.KEYPAIR:
            raise InvalidAPIParameters("Unsupported authorization type")
        auth_config = self._config_provider.config.auth
        login_client_type_id = action.client_type_id
        user, active_sessions = await self._verify_user(action, auth_config, login_client_type_id)

        try:
            post_result = await self._post_check(action, user, active_sessions, auth_config)
            if isinstance(post_result, AuthorizeActionResult):
                return post_result
            keypair_row, live_sessions = post_result

            return await self._create_login_session(
                action, user, keypair_row, live_sessions, auth_config, login_client_type_id
            )
        except (
            AuthorizationFailed,
            PasswordExpired,
            RejectedByHook,
            TooManyConcurrentLoginSessions,
        ) as e:
            await self._record_login_failure(
                user.uuid,
                action.domain_name,
                _classify_failure(e),
            )
            raise

    async def _verify_user(
        self,
        action: AuthorizeAction,
        auth_config: AuthConfig,
        login_client_type_id: uuid.UUID | None,
    ) -> tuple[RowMapping, list[ActiveSessionInfo]]:
        """Step 1: Verify user identity via hook or password."""
        params = action.hook_params
        hook_result = await self._hook_plugin_ctx.dispatch(
            "AUTHORIZE",
            (action.request, params),
            return_when=FIRST_COMPLETED,
        )
        if hook_result.status != PASSED:
            raise RejectedByHook.from_hook_result(hook_result)
        if hook_result.result:
            user = hook_result.result
            active_sessions = await self._auth_repository.get_active_session_tokens(
                user.uuid, login_client_type_id=login_client_type_id
            )
            return user, active_sessions

        target_password_info = PasswordInfo(
            password=action.password,
            algorithm=auth_config.password_hash_algorithm,
            rounds=auth_config.password_hash_rounds,
            salt_size=auth_config.password_hash_salt_size,
        )
        cred_result = await self._auth_repository.verify_credential(
            action.domain_name,
            action.email,
            target_password_info=target_password_info,
            login_client_type_id=login_client_type_id,
        )
        return cred_result.user, cred_result.active_sessions

    async def _post_check(
        self,
        action: AuthorizeAction,
        user: RowMapping,
        active_sessions: list[ActiveSessionInfo],
        auth_config: AuthConfig,
    ) -> tuple[Any, list[ActiveSessionInfo]] | AuthorizeActionResult:
        """Step 2: User status checks, keypair lookup, POST_AUTHORIZE hook, Valkey cross-check."""
        if user.status == UserStatus.BEFORE_VERIFICATION:
            raise AuthorizationFailed("This account needs email verification.")
        if user.status in INACTIVE_USER_STATUSES:
            raise AuthorizationFailed("User credential mismatch.")
        await self._check_password_age(user, auth_config)

        user_row = await self._auth_repository.get_user_row_by_uuid(user.uuid)
        default_keypair_row = user_row.get_default_keypair_row()
        if default_keypair_row is None:
            raise AuthorizationFailed("No API keypairs found.")

        hook_result = await self._hook_plugin_ctx.dispatch(
            "POST_AUTHORIZE",
            (action.request, action.hook_params, user, default_keypair_row.mapping),
            return_when=FIRST_COMPLETED,
        )
        if hook_result.status != PASSED:
            raise RejectedByHook.from_hook_result(hook_result)
        if hook_result.result is not None and isinstance(hook_result.result, web.StreamResponse):
            return AuthorizeActionResult(
                stream_response=hook_result.result,
                authorization_result=None,
            )

        # Cross-check active sessions with Valkey
        live_sessions: list[ActiveSessionInfo] = []
        for session_info in active_sessions:
            if await self._valkey_session_client.get_login_session(session_info.session_token):
                live_sessions.append(session_info)
            else:
                await self._auth_repository.delete_login_session_by_token(
                    session_info.session_token, LoginAttemptResult.EXPIRED
                )

        return default_keypair_row, live_sessions

    def _enforce_max_concurrent_logins(
        self,
        max_concurrent_logins: int | None,
        live_sessions: list[ActiveSessionInfo],
        force: bool,
    ) -> list[str] | None:
        """Apply the per-user ``max_concurrent_logins`` cap and return tokens to evict.

        Behavior:
        - ``max_concurrent_logins is None`` (unlimited or no policy): no cap, returns ``None``.
        - Below the cap: no eviction, returns ``None``.
        - At or over the cap with ``force=True``: returns the oldest session tokens
          (``count - max_concurrent_logins + 1`` of them) so the caller can invalidate
          them before creating the new login session.
        - At or over the cap with ``force=False``: raises ``TooManyConcurrentLoginSessions``.

        ``live_sessions`` must already be the authoritative active set (cross-checked
        against Valkey by the caller) and ordered oldest-first.
        """
        if max_concurrent_logins is None:
            return None
        if max_concurrent_logins <= 0:
            # Cap of 0 (or negative) means no logins allowed at all; force cannot help
            # since even evicting every existing session still leaves the new login over cap.
            raise TooManyConcurrentLoginSessions()
        count = len(live_sessions)
        if count < max_concurrent_logins:
            return None
        if not force:
            raise TooManyConcurrentLoginSessions()
        # Normally evicts just the oldest session (result is 1). Can be >1 when an admin
        # lowered max_concurrent_logins after the user already exceeded the new cap — in
        # that case this brings the user back down to (max_concurrent_logins - 1) in one shot.
        sessions_to_invalidate = count - max_concurrent_logins + 1
        return [s.session_token for s in live_sessions[:sessions_to_invalidate]]

    async def _create_login_session(
        self,
        action: AuthorizeAction,
        user: RowMapping,
        keypair_row: Any,
        live_sessions: list[ActiveSessionInfo],
        auth_config: AuthConfig,
        login_client_type_id: uuid.UUID | None,
    ) -> AuthorizeActionResult:
        """Step 3: Create login session (DB + Valkey), force-invalidate old sessions if needed.

        Enforcement uses ``user_resource_policy.max_concurrent_logins``:
        - None (unlimited): always proceed.
        - Set: if ``len(live_sessions) >= limit`` and ``action.force`` is True, evict the
          oldest sessions to make room; otherwise raise ``TooManyConcurrentLoginSessions``.

        ``live_sessions`` is already the authoritative active set (cross-checked against
        Valkey by the caller), so no additional repository count query is needed.
        """
        try:
            user_resource_policy = await self._user_resource_policy_repository.get_by_name(
                user.resource_policy
            )
            max_concurrent_logins: int | None = user_resource_policy.max_concurrent_logins
        except UserResourcePolicyNotFound:
            # If no matching resource policy is found, skip login-session limit enforcement.
            max_concurrent_logins = None

        tokens_to_invalidate = self._enforce_max_concurrent_logins(
            max_concurrent_logins=max_concurrent_logins,
            live_sessions=live_sessions,
            force=action.force,
        )

        # Create-before-destroy: evict old sessions only after the new one is persisted.
        session_result = await self._auth_repository.create_login_session(
            user_id=user.uuid,
            access_key=keypair_row.access_key,
            domain_name=action.domain_name,
            login_client_type_id=login_client_type_id,
        )

        if tokens_to_invalidate:
            for token in tokens_to_invalidate:
                await self._valkey_session_client.delete_login_session(token)
            await self._auth_repository.delete_login_sessions_by_tokens(
                tokens_to_invalidate, LoginAttemptResult.EVICTED
            )

        session_data = LoginSessionData(
            created=int(time.time()),
            expiration_dt=int(time.time()) + auth_config.login_session_max_age,
            session=LoginSessionInner(
                authenticated=True,
                token=LoginSessionTokenData(
                    type="keypair",
                    access_key=keypair_row.access_key,
                    secret_key=keypair_row.secret_key,
                    role=user.role,
                    status=user.status,
                ),
            ),
        )
        await self._valkey_session_client.set_login_session(
            session_result.session_token,
            session_data.model_dump_json(),
            auth_config.login_session_max_age,
        )

        return AuthorizeActionResult(
            stream_response=None,
            authorization_result=AuthorizationResult(
                access_key=AccessKey(keypair_row.access_key),
                secret_key=SecretKey(keypair_row.secret_key),
                user_id=UserID(user.uuid),
                role=UserRole(user.role),
                status=user.status,
                session_token=session_result.session_token,
            ),
        )

    async def _record_login_failure(
        self,
        user_uuid: uuid.UUID,
        domain_name: str,
        result: LoginAttemptResult,
    ) -> None:
        try:
            await self._auth_repository.record_login_history(user_uuid, domain_name, result)
        except Exception:
            log.warning("Failed to record login history: {} for user {}", result, user_uuid)

    async def signup(self, action: SignupAction) -> SignupActionResult:
        params = action.hook_params
        hook_result = await self._hook_plugin_ctx.dispatch(
            "PRE_SIGNUP",
            (params,),
            return_when=ALL_COMPLETED,
        )
        if hook_result.status != PASSED:
            raise RejectedByHook.from_hook_result(hook_result)
        # Merge the hook results as a single map.
        hook_results = cast(list[Mapping[str, Any]], hook_result.result or [])
        # Convert Mapping to dict for ChainMap compatibility
        user_data_overriden: ChainMap[str, Any] = ChainMap(*[
            dict(result) for result in hook_results
        ])

        # [Hooking point for VERIFY_PASSWORD_FORMAT with the ALL_COMPLETED requirement]
        # The hook handlers should accept the request and whole ``params` dict.
        # They should return None if the validation is successful and raise the
        # Reject error otherwise.
        hook_result = await self._hook_plugin_ctx.dispatch(
            "VERIFY_PASSWORD_FORMAT",
            (action.request, params),
            return_when=ALL_COMPLETED,
        )
        if hook_result.status != PASSED:
            hook_result.reason = hook_result.reason or "invalid password format"
            raise RejectedByHook.from_hook_result(hook_result)

        # Check if email already exists.
        if await self._auth_repository.check_email_exists(action.email):
            raise EmailAlreadyExistsError("Email already exists")

        # Create a user.
        # Create PasswordInfo for the new user's password
        auth_config = self._config_provider.config.auth
        password_info = PasswordInfo(
            password=action.password,
            algorithm=auth_config.password_hash_algorithm,
            rounds=auth_config.password_hash_rounds,
            salt_size=auth_config.password_hash_salt_size,
        )

        user_spec = UserCreatorSpec(
            domain_name=action.domain_name,
            username=action.username if action.username is not None else action.email,
            email=action.email,
            password=password_info,
            need_password_change=False,
            full_name=action.full_name if action.full_name is not None else "",
            description=action.description if action.description is not None else "",
            status=UserStatus.INACTIVE,
            status_info="user-signup",
            role=UserRole.USER,
            resource_policy="default",
            sudo_session_enabled=False,
        )
        if user_data_overriden:
            spec_fields = {f.name for f in dataclasses.fields(UserCreatorSpec)}
            overrides = {
                # Hooks name the DB column; the spec field is integration_name.
                ("integration_name" if key == "integration_id" else key): val
                for key, val in user_data_overriden.items()
                if key != "resource_policy"  # resource_policy in user_data is for keypair
            }
            user_spec = dataclasses.replace(
                user_spec,
                **{key: val for key, val in overrides.items() if key in spec_fields},
            )

        # Resolve the default project to enroll the new user in.
        group_name = user_data_overriden.get("group", DEFAULT_PROJECT_NAME)
        project_id = await self._group_repository.project_id_by_name_in_domain(
            action.domain_name, group_name
        )

        resource_policy = user_data_overriden.get("resource_policy", "default")
        creation = await self._auth_repository.create_user_with_keypair(
            user_spec=user_spec,
            project_ids=[project_id] if project_id is not None else [],
            keypair_resource_policy=resource_policy,
            keypair_rate_limit=1000,
        )
        user = creation.user
        keypair = creation.keypair

        # [Hooking point for POST_SIGNUP as one-way notification]
        # The hook handlers should accept a tuple of the user email,
        # the new user's UUID, and a dict with initial user's preferences.
        initial_user_prefs = {
            "lang": action.request.headers.get("Accept-Language", "en-us").split(",")[0].lower(),
        }
        await self._hook_plugin_ctx.notify(
            "POST_SIGNUP",
            (action.email, user.uuid, initial_user_prefs),
        )
        return SignupActionResult(
            user_id=user.uuid,
            access_key=keypair.access_key,
            secret_key=keypair.secret_key,
        )

    async def logout(self, action: LogoutAction) -> LogoutActionResult:
        await self._auth_repository.delete_login_session_by_token(
            action.session_token, LoginAttemptResult.LOGOUT
        )
        await self._valkey_session_client.delete_login_session(action.session_token)
        return LogoutActionResult(success=True)

    async def admin_revoke_login_session(
        self, action: AdminRevokeLoginSessionAction
    ) -> RevokeLoginSessionActionResult:
        session_token = await self._auth_repository.delete_login_session_by_id(
            action.session_id, LoginAttemptResult.REVOKED_BY_ADMIN
        )
        await self._valkey_session_client.delete_login_session(session_token)
        return RevokeLoginSessionActionResult(success=True)

    async def my_revoke_login_session(
        self, action: MyRevokeLoginSessionAction
    ) -> RevokeLoginSessionActionResult:
        session_data = await self._auth_repository.get_login_session_by_id(action.session_id)
        if session_data.user_id != action.user_id:
            raise GenericForbidden("You can only revoke your own login sessions.")
        session_token = await self._auth_repository.delete_login_session_by_id(
            action.session_id, LoginAttemptResult.REVOKED_BY_USER
        )
        await self._valkey_session_client.delete_login_session(session_token)
        return RevokeLoginSessionActionResult(success=True)

    async def admin_unblock_user(
        self, action: AdminUnblockUserAction
    ) -> AdminUnblockUserActionResult:
        await self._valkey_session_client.clear_login_block(action.username)
        return AdminUnblockUserActionResult(success=True)

    async def signout(self, action: SignoutAction) -> SignoutActionResult:
        if action.email != action.requester_email:
            raise GenericForbidden("Not the account owner")
        email = action.email
        await self._auth_repository.check_credential_without_migration(
            action.domain_name,
            email,
            action.password,
        )
        deleted_tokens = await self._auth_repository.delete_user_login_sessions(
            action.user_id, action.domain_name, LoginAttemptResult.LOGOUT
        )
        for token in deleted_tokens:
            await self._valkey_session_client.delete_login_session(token)
        await self._auth_repository.deactivate_user_and_keypairs(email)

        return SignoutActionResult(success=True)

    async def update_full_name(self, action: UpdateFullNameAction) -> UpdateFullNameActionResult:
        await self._auth_repository.update_user_full_name(
            action.email, action.domain_name, action.full_name
        )
        return UpdateFullNameActionResult(success=True)

    async def update_password(self, action: UpdatePasswordAction) -> UpdatePasswordActionResult:
        domain_name = action.domain_name
        email = action.email
        log_fmt = "AUTH.UPDATE_PASSWORD(d:{}, email:{})"
        log_args = (domain_name, email)
        if action.new_password != action.new_password_confirm:
            log.info(log_fmt + ": new password mismtach", *log_args)
            return UpdatePasswordActionResult(
                success=False,
                message="new password mismatch",
            )
        try:
            await self._auth_repository.check_credential_without_migration(
                domain_name,
                email,
                action.old_password,
            )
        except AuthorizationFailed as e:
            log.info(log_fmt + ": old password mismatch", *log_args)
            raise AuthorizationFailed("Old password mismatch") from e

        # [Hooking point for VERIFY_PASSWORD_FORMAT with the ALL_COMPLETED requirement]
        # The hook handlers should accept the request and whole ``params` dict.
        # They should return None if the validation is successful and raise the
        # Reject error otherwise.
        hook_result = await self._hook_plugin_ctx.dispatch(
            "VERIFY_PASSWORD_FORMAT",
            (action.request, action.hook_params),
            return_when=ALL_COMPLETED,
        )
        if hook_result.status != PASSED:
            hook_result.reason = hook_result.reason or "invalid password format"
            raise RejectedByHook.from_hook_result(hook_result)

        # Create PasswordInfo with config values
        auth_config = self._config_provider.config.auth
        password_info = PasswordInfo(
            password=action.new_password,
            algorithm=auth_config.password_hash_algorithm,
            rounds=auth_config.password_hash_rounds,
            salt_size=auth_config.password_hash_salt_size,
        )
        await self._auth_repository.update_user_password(email, password_info)

        return UpdatePasswordActionResult(
            success=True,
            message="Password updated successfully",
        )

    async def update_password_no_auth(
        self, action: UpdatePasswordNoAuthAction
    ) -> UpdatePasswordNoAuthActionResult:
        auth_config = self._config_provider.config.auth
        if auth_config.max_password_age is None:
            raise GenericBadRequest("Unsupported function.")
        checked_user = await self._auth_repository.check_credential_without_migration(
            action.domain_name,
            action.email,
            password=action.current_password,
        )
        new_password = action.new_password
        if compare_to_hashed_password(new_password, checked_user["password"]):
            raise AuthorizationFailed("Cannot update to the same password as an existing password.")

        # [Hooking point for VERIFY_PASSWORD_FORMAT with the ALL_COMPLETED requirement]
        # The hook handlers should accept the request and whole ``params` dict.
        # They should return None if the validation is successful and raise the
        # Reject error otherwise.
        hook_result = await self._hook_plugin_ctx.dispatch(
            "VERIFY_PASSWORD_FORMAT",
            (action.request, action.hook_params),
            return_when=ALL_COMPLETED,
        )
        if hook_result.status != PASSED:
            hook_result.reason = hook_result.reason or "invalid password format"
            raise RejectedByHook.from_hook_result(hook_result)

        password_info = PasswordInfo(
            password=new_password,
            algorithm=auth_config.password_hash_algorithm,
            rounds=auth_config.password_hash_rounds,
            salt_size=auth_config.password_hash_salt_size,
        )
        changed_at = await self._auth_repository.update_user_password_by_uuid(
            checked_user["uuid"], password_info
        )

        return UpdatePasswordNoAuthActionResult(
            user_id=checked_user["uuid"],
            password_changed_at=changed_at,
        )

    async def get_ssh_keypair(self, action: GetSSHKeypairAction) -> GetSSHKeypairActionResult:
        pubkey = await self._auth_repository.get_ssh_public_key(action.access_key)
        return GetSSHKeypairActionResult(public_key=pubkey or "", access_key=action.access_key)

    async def generate_ssh_keypair(
        self, action: GenerateSSHKeypairAction
    ) -> GenerateSSHKeypairActionResult:
        pubkey, privkey = generate_ssh_keypair()
        await self._auth_repository.update_ssh_keypair(action.access_key, pubkey, privkey)

        return GenerateSSHKeypairActionResult(
            ssh_keypair=SSHKeypair(
                ssh_public_key=pubkey,
                ssh_private_key=privkey,
            ),
            user_id=action.user_id,
        )

    async def upload_ssh_keypair(
        self, action: UploadSSHKeypairAction
    ) -> UploadSSHKeypairActionResult:
        privkey = action.private_key
        pubkey = action.public_key
        self._ssh_key_validator.validate(SSHPrivateKey(privkey), SSHPublicKey(pubkey))

        await self._auth_repository.update_ssh_keypair(action.access_key, pubkey, privkey)

        return UploadSSHKeypairActionResult(
            ssh_keypair=SSHKeypair(
                ssh_public_key=pubkey,
                ssh_private_key=privkey,
            ),
            user_id=action.user_id,
        )

    async def resolve_access_key_scope(
        self, action: ResolveAccessKeyScopeAction
    ) -> ResolveAccessKeyScopeResult:
        requester_ak = AccessKey(action.requester_access_key)
        if (
            action.owner_access_key is None
            or action.owner_access_key == action.requester_access_key
        ):
            return ResolveAccessKeyScopeResult(
                requester_access_key=requester_ak,
                owner_access_key=requester_ak,
            )
        owner_ak = AccessKey(action.owner_access_key)
        try:
            (
                owner_domain,
                owner_role,
            ) = await self._auth_repository.get_delegation_target_by_access_key(
                action.owner_access_key,
            )
        except ValueError as e:
            raise InvalidAPIParameters(str(e)) from e
        try:
            check_if_requester_is_eligible_to_act_as_target_user(
                action.requester_role,
                action.requester_domain,
                owner_role,
                owner_domain,
            )
        except RuntimeError as e:
            raise GenericForbidden(str(e)) from e
        return ResolveAccessKeyScopeResult(
            requester_access_key=requester_ak,
            owner_access_key=owner_ak,
        )

    async def resolve_user_id_by_access_key(
        self, action: ResolveUserIDByAccessKeyAction
    ) -> ResolveUserIDByAccessKeyResult:
        user_id = await self._auth_repository.get_user_id_by_access_key(action.access_key)
        return ResolveUserIDByAccessKeyResult(user_id=user_id)

    async def resolve_user_scope(self, action: ResolveUserScopeAction) -> ResolveUserScopeResult:
        if action.owner_user_email is None:
            return ResolveUserScopeResult(
                owner_uuid=action.requester_uuid,
                owner_role=action.requester_role,
            )
        if not action.is_superadmin:
            raise InvalidAPIParameters("Only superadmins may have user scopes.")
        try:
            (
                owner_uuid,
                owner_role,
                owner_domain,
            ) = await self._auth_repository.get_delegation_target_by_email(
                action.owner_user_email,
            )
        except ValueError as e:
            raise InvalidAPIParameters(str(e)) from e
        try:
            check_if_requester_is_eligible_to_act_as_target_user(
                action.requester_role,
                action.requester_domain,
                owner_role,
                owner_domain,
            )
        except RuntimeError as e:
            raise GenericForbidden(str(e)) from e
        return ResolveUserScopeResult(
            owner_uuid=owner_uuid,
            owner_role=owner_role,
        )

    async def admin_search_login_sessions(
        self, action: AdminSearchLoginSessionsAction
    ) -> SearchLoginSessionsActionResult:
        result = await self._auth_repository.admin_search_login_sessions(querier=action.querier)
        return SearchLoginSessionsActionResult(result=result)

    async def search_login_sessions(
        self, action: SearchLoginSessionsAction
    ) -> SearchLoginSessionsActionResult:
        result = await self._auth_repository.search_login_sessions(
            scope=action.scope, querier=action.querier
        )
        return SearchLoginSessionsActionResult(result=result)

    async def admin_search_login_history(
        self, action: AdminSearchLoginHistoryAction
    ) -> SearchLoginHistoryActionResult:
        result = await self._auth_repository.admin_search_login_history(querier=action.querier)
        return SearchLoginHistoryActionResult(result=result)

    async def search_login_history(
        self, action: SearchLoginHistoryAction
    ) -> SearchLoginHistoryActionResult:
        result = await self._auth_repository.search_login_history(
            scope=action.scope, querier=action.querier
        )
        return SearchLoginHistoryActionResult(result=result)

    async def _check_password_age(self, user: RowMapping, auth_config: AuthConfig | None) -> None:
        if (
            auth_config is not None
            and (max_password_age := auth_config.max_password_age) is not None
        ):
            password_changed_at: datetime | None = user.password_changed_at
            if password_changed_at is None:
                return  # Skip check if password_changed_at is not set

            current_dt: datetime = await self._auth_repository.get_current_time()
            if password_changed_at + max_password_age < current_dt:
                # Force user to update password
                raise PasswordExpired(
                    extra_msg=f"Password expired on {password_changed_at + max_password_age}."
                )