fix: #1641 Allow delete inactive users (#1661)

server/admin_management/api/app/routers/router_guards.py

@@ -63,7 +63,7 @@ def authorize_by_fam_admin(claims: dict = Depends(validate_token)):
 
 # for app admin and FAM admin, we require the access group in the token
 # the get_request_cognito_user_id will check the access in the token
-async def get_current_requester(
+def get_current_requester(
     request_cognito_user_id: str = Depends(get_request_cognito_user_id),
     access_roles=Depends(get_access_roles),
     user_service: UserService = Depends(user_service_instance),
@@ -83,7 +83,7 @@ async def get_current_requester(
 # for delegated admin, there is no access group in the token, our auth lambda function only add app admin to the token
 # get_request_cognito_user_id_without_access_check will return the requester without checking the access group in the token
 # this should only used by the get_admin_user_access API, all other APIs require admin access in the token
-async def get_current_requester_without_access_check(
+def get_current_requester_without_access_check(
     request_cognito_user_id: str = Depends(
         get_request_cognito_user_id_without_access_check
     ),
@@ -203,7 +203,7 @@ async def get_request_role_from_id(
             return None
 
 
-async def get_verified_target_user(
+def get_verified_target_user(
     requester: Requester = Depends(get_current_requester),
     target_user: TargetUser = Depends(get_target_user_from_id),
     role: FamRole = Depends(get_request_role_from_id),
@@ -222,7 +222,7 @@ async def get_verified_target_user(
     return target_user_validator.verify_user_exist()
 
 
-async def enforce_self_grant_guard(
+def enforce_self_grant_guard(
     requester: Requester = Depends(get_current_requester),
     target_user: TargetUser = Depends(get_target_user_from_id),
 ):
@@ -297,7 +297,7 @@ def authorize_by_app_id(
         )
 
 
-async def validate_param_application_admin_id(
+def validate_param_application_admin_id(
     application_admin_id: int,
     application_admin_service: ApplicationAdminService = Depends(
         application_admin_service_instance
@@ -313,7 +313,7 @@ async def validate_param_application_admin_id(
         )
 
 
-async def validate_param_application_id(
+def validate_param_application_id(
     application_admin_request: FamAppAdminCreateRequest,
     application_service: ApplicationService = Depends(application_service_instance),
 ):
@@ -329,7 +329,7 @@ async def validate_param_application_id(
         )
 
 
-async def validate_param_user_type(application_admin_request: FamAppAdminCreateRequest):
+def validate_param_user_type(application_admin_request: FamAppAdminCreateRequest):
     if (
         not application_admin_request.user_type_code
         or application_admin_request.user_type_code != UserType.IDIR
@@ -342,7 +342,7 @@ async def validate_param_user_type(application_admin_request: FamAppAdminCreateR
         )
 
 
-async def validate_param_access_control_privilege_id(
+def validate_param_access_control_privilege_id(
     access_control_privilege_id: int,
     access_control_privilege_service: AccessControlPrivilegeService = Depends(
         access_control_privilege_service_instance

server/backend/api/app/crud/validator/target_user_validator.py

@@ -1,19 +1,13 @@
 import copy
 import logging
 
-from api.app.constants import (
-    ERROR_CODE_INVALID_REQUEST_PARAMETER,
-    ApiInstanceEnv,
-    IdimSearchUserParamType,
-    UserType,
-)
+from api.app.constants import (ERROR_CODE_INVALID_REQUEST_PARAMETER,
+                               ApiInstanceEnv, IdimSearchUserParamType,
+                               UserType)
 from api.app.integration.idim_proxy import IdimProxyService
-from api.app.schemas import (
-    IdimProxyBceidSearchParamSchema,
-    IdimProxySearchParamSchema,
-    RequesterSchema,
-    TargetUserSchema,
-)
+from api.app.schemas import (IdimProxyBceidSearchParamSchema,
+                             IdimProxySearchParamSchema, RequesterSchema,
+                             TargetUserSchema)
 from api.app.utils import utils
 
 LOGGER = logging.getLogger(__name__)
@@ -26,7 +20,7 @@ def __init__(
         target_user: TargetUserSchema,
         api_instance_env: ApiInstanceEnv,
     ):
-        LOGGER.debug(f"Validating target env set to: {api_instance_env}")
+        LOGGER.debug(f"Validating target_user - {target_user.user_id}, target env set to: {api_instance_env}")
         self.verified_target_user = copy.deepcopy(target_user)
         self.idim_proxy_service = IdimProxyService(requester, api_instance_env)
 
@@ -96,8 +90,10 @@ def verify_user_exist(self) -> TargetUserSchema:
             self.verified_target_user.email = search_result.get("email")
 
         else:
-            error_msg = f"Invalid request, cannot find user {self.verified_target_user.user_name} "
-            f"{self.verified_target_user.user_guid} with user type {self.verified_target_user.user_type_code}"
+            error_msg = (
+                f"Invalid request, cannot find user {self.verified_target_user.user_name} "
+                f"{self.verified_target_user.user_guid} with user type {self.verified_target_user.user_type_code}"
+            )
             utils.raise_http_exception(
                 error_code=ERROR_CODE_INVALID_REQUEST_PARAMETER,
                 error_msg=error_msg,

server/backend/api/app/routers/router_guards.py

@@ -12,8 +12,8 @@
                                ERROR_CODE_MISSING_KEY_ATTRIBUTE,
                                ERROR_CODE_REQUESTER_NOT_EXISTS,
                                ERROR_CODE_SELF_GRANT_PROHIBITED,
-                               ERROR_CODE_TERMS_CONDITIONS_REQUIRED, RoleType,
-                               UserType)
+                               ERROR_CODE_TERMS_CONDITIONS_REQUIRED,
+                               ERROR_CODE_UNKNOWN_STATE, RoleType, UserType)
 from api.app.crud import (crud_access_control_privilege, crud_role, crud_user,
                           crud_user_role, crud_utils)
 from api.app.crud.validator.target_user_validator import TargetUserValidator
@@ -318,7 +318,7 @@ async def get_target_user_from_id(
         return target_new_user
 
 
-async def authorize_by_user_type(
+def authorize_by_user_type(
     requester: RequesterSchema = Depends(get_current_requester),
     target_user: TargetUserSchema = Depends(get_target_user_from_id),
 ):
@@ -344,7 +344,7 @@ async def authorize_by_user_type(
             )
 
 
-async def internal_only_action(
+def internal_only_action(
     requester: RequesterSchema = Depends(get_current_requester),
 ):
     if requester.user_type_code is not UserType.IDIR:
@@ -366,7 +366,7 @@ def external_delegated_admin_only_action(
         )
 
 
-async def enforce_self_grant_guard(
+def enforce_self_grant_guard(
     requester: RequesterSchema = Depends(get_current_requester),
     target_user: TargetUserSchema = Depends(get_target_user_from_id),
 ):
@@ -392,7 +392,7 @@ async def enforce_self_grant_guard(
         )
 
 
-async def get_verified_target_user(
+def get_verified_target_user(
     requester: RequesterSchema = Depends(get_current_requester),
     target_user: TargetUserSchema = Depends(get_target_user_from_id),
     role: FamRole = Depends(get_request_role_from_id),
@@ -412,21 +412,46 @@ async def enforce_bceid_by_same_org_guard(
     # forbid business bceid user (requester) manage idir user's access
     _enforce_user_type_auth: None = Depends(authorize_by_user_type),
     requester: RequesterSchema = Depends(get_current_requester),
-    target_user: TargetUserSchema = Depends(get_verified_target_user),
+    target_user: TargetUserSchema = Depends(get_target_user_from_id),
+    role: FamRole = Depends(get_request_role_from_id)
 ):
     """
-    When requester is a BCeID user, enforce requester can only manage target
-    user from the same organization.
-    :param _enforce_user_type_auth: call the authorize_by_user_type method
-            to ensure that this enforce_bceid_by_same_org_guard method only
-            checks when business bceid user grants access to another business
-            bceid user
+    When requester is a BCeID user, enforce requester can only manage target user from the same organization.
+    """
+    """
+    Initially this guard was using dependency validation: "Depends(get_verified_target_user)" to retrieve
+    information from IDP(IDIM service) for this guard for checking.
+    But due to special case for - deleting user/role assignment when target user is no longer available from IDP
+    (inactive or removed for some reason), we still like to be able to delete the target user from db, however,
+    we won't be able to verify user from IDIM in this case.
+    For this case we won't be able to use "get_verified_target_user" as a guard; moving it from router guard
+    dependency to be in this method's body for special handling. So:
+
+    - For IDIR requester, there is no need to get "verified target user" as IDIR user can remove any IDIR user
+      and any BCeID user.
+
+    - For BCeID requester, get "verified target user" is still needed for BCeID target user to check if they
+      are from the same organization. But, in the case if target user cannot be verified, then raise exception
+      for this case.
     """
     LOGGER.debug(
         f"Verifying requester {requester.user_name} (type {requester.user_type_code}) "
         "is allowed to manage BCeID target user for the same organization."
     )
+
     if requester.user_type_code == UserType.BCEID:
+        try:
+            target_user = get_verified_target_user(requester=requester, target_user=target_user, role=role)
+        except Exception as e:
+            LOGGER.error(f"Target user could not be verified: {e}.")
+            app_name = role.application.application_name
+            error_msg = f"Unable to verify user {target_user.user_name}. Please contact {app_name} administrator for the action."
+            utils.raise_http_exception(
+                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+                error_code=ERROR_CODE_UNKNOWN_STATE,
+                error_msg=error_msg,
+            )
+
         requester_business_guid = requester.business_guid
         target_user_business_guid = target_user.business_guid
 

server/backend/testspg/conftest.py

@@ -191,8 +191,9 @@ def _get_current_requester_by_token(access_token: str) -> RequesterSchema:
 
 
 @pytest.fixture(scope="function")
-def override_get_verified_target_user(test_client_fixture):
-    # mock the return result for idim validation of the target user, to avoid calling external idim-proxy
+def override_depends__get_verified_target_user(test_client_fixture):
+    # Override FastAPI dependency "get_verified_target_user".
+    # Mock the return result for idim validation of the target user, to avoid calling external idim-proxy
     def _override_get_verified_target_user(mocked_data=ACCESS_GRANT_FOM_DEV_CR_IDIR):
         app = test_client_fixture.app
         app.dependency_overrides[get_verified_target_user] = lambda: TargetUserSchema(
@@ -202,9 +203,36 @@ def _override_get_verified_target_user(mocked_data=ACCESS_GRANT_FOM_DEV_CR_IDIR)
     return _override_get_verified_target_user
 
 
+@pytest.fixture(scope="function")
+def mock_verified_target_user(mocker):
+    """
+    This fixture has the same purpose but different than similar fixture 'override_depends__get_verified_target_user'
+    on testing usage.
+    It mocks the return result from idim validation of the target user. However, this is not overriding the FastAPI
+    dependency.
+    For example:
+    Delete user/role assignment related to "get_verified_target_user" is a special case (not as FastAPI dependency),
+    and needs to be mocked individually at function.
+    """
+
+    def _mock_verified_target_user(
+        mocked_user: Optional[TargetUserSchema] = None,
+        mocked_side_effect: Optional[Exception] = None
+    ):
+        patch_fn_path = "api.app.routers.router_guards.get_verified_target_user"
+        if mocked_user:
+            return mocker.patch(patch_fn_path, return_value=mocked_user)
+        elif mocked_side_effect:
+            return mocker.patch(patch_fn_path, side_effect=mocked_side_effect)
+        else:
+            raise Exception("Programming Error. Missing arguments")
+
+    return _mock_verified_target_user
+
+
 @pytest.fixture(scope="function")
 def create_test_user_role_assignments(
-    test_client_fixture, override_get_verified_target_user
+    test_client_fixture, override_depends__get_verified_target_user
 ):
     """
     Convenient function to assign multipe users to an application.
@@ -217,7 +245,7 @@ def create_test_user_role_assignments(
     def _create_test_user_role_assignments(requester_token, request_bodies: List[dict]):
         created_users = []
         for request_body in request_bodies:
-            override_get_verified_target_user(request_body)
+            override_depends__get_verified_target_user(request_body)
             created_users.append(
                 create_test_user_role_assignment(
                     test_client_fixture=test_client_fixture,
@@ -245,7 +273,7 @@ def create_test_user_role_assignment(
 
 
 @pytest.fixture(scope="function")
-def override_enforce_bceid_terms_conditions_guard(test_client_fixture):
+def override_depends__enforce_bceid_terms_conditions_guard(test_client_fixture):
     # Override T&C checks based on test cases scenarios.
     def _override_enforce_bceid_terms_conditions_guard(mocked_tc_accepted=True):
         app = test_client_fixture.app
@@ -322,6 +350,16 @@ def mock_forest_client_integration_service():
         yield m.return_value  # Very important to get instance of mocked class.
 
 
+@pytest.fixture(scope="function", autouse=True)
+def mock_idim_proxy_integratioin_service():
+    # Mocked dependency class object
+    with patch(
+        "api.app.integration.idim_proxy.IdimProxyService",
+        autospec=True,
+    ) as m:
+        yield m.return_value  # Very important to get instance of mocked class.
+
+
 @pytest.fixture(scope="function")
 def default_app_role_assignment_page_Params() -> UserRolePageParamsSchema:
     return UserRolePageParamsSchema(
@@ -330,4 +368,4 @@ def default_app_role_assignment_page_Params() -> UserRolePageParamsSchema:
         search=None,
         sort_by=None,
         sort_order=None
-    )
+    )

server/backend/testspg/crud/test_idim_proxy_service.py

@@ -1,27 +1,22 @@
-import os
 import copy
 import logging
+import os
 
 import pytest
 from api.app.constants import IdimSearchUserParamType
 from api.app.integration.idim_proxy import IdimProxyService
 from api.app.jwt_validation import ERROR_PERMISSION_REQUIRED
-from api.app.schemas import (
-    IdimProxyBceidSearchParamSchema,
-    IdimProxySearchParamSchema,
-    RequesterSchema,
-)
+from api.app.schemas import (IdimProxyBceidSearchParamSchema,
+                             IdimProxySearchParamSchema, RequesterSchema)
 from fastapi import HTTPException
 from requests import HTTPError
-from testspg.constants import (
-    TEST_BCEID_REQUESTER_DICT,
-    TEST_IDIR_REQUESTER_DICT,
-    USER_GUID_BCEID_LOAD_3_TEST,
-    USER_GUID_BCEID_LOAD_3_TEST_CHILD_1,
-    USER_NAME_BCEID_LOAD_2_TEST,
-    USER_NAME_BCEID_LOAD_3_TEST,
-    USER_NAME_BCEID_LOAD_3_TEST_CHILD_1,
-)
+from testspg.constants import (TEST_BCEID_REQUESTER_DICT,
+                               TEST_IDIR_REQUESTER_DICT,
+                               USER_GUID_BCEID_LOAD_3_TEST,
+                               USER_GUID_BCEID_LOAD_3_TEST_CHILD_1,
+                               USER_NAME_BCEID_LOAD_2_TEST,
+                               USER_NAME_BCEID_LOAD_3_TEST,
+                               USER_NAME_BCEID_LOAD_3_TEST_CHILD_1)
 
 LOGGER = logging.getLogger(__name__)
 
@@ -110,14 +105,24 @@ def test_search_idir__user_not_exist_no_user_found(self):
 
     # --- Performs search_business_bceid user (IDIR Requester/BCeID Requester) ---
 
-    def test_search_bceid__user_not_exist_not_found(self):
+    def test_search_bceid__idir_requester_user_not_exist_not_found(self):
         idim_proxy_api = IdimProxyService(self.requester_idir)
         search_params = copy.deepcopy(self.search_params_business_bceid_same_org)
-        not_exists_idir_user = "USERNOTEXISTS"
-        search_params.searchValue = not_exists_idir_user
+        search_params.searchValue = "USERNOTEXISTS"
+        search_result = idim_proxy_api.search_business_bceid(search_params)
+
+        assert search_result["found"] == False
+        assert search_result["userId"] == search_params.searchValue
+
+    def test_search_bceid__bceid_requester_user_not_exist_not_found(self):
+        # test bceid search for unknow bceid
+        idim_proxy_api = IdimProxyService(self.requester_business_bceid)
+        search_params = copy.deepcopy(self.search_params_business_bceid_same_org)
+        search_params.searchValue = "USERNOTEXISTS"
         search_result = idim_proxy_api.search_business_bceid(search_params)
 
         assert search_result["found"] == False
+        assert search_result["userId"] == search_params.searchValue
 
     def test_search_bceid__idir_requester_by_userid_search_pass(self):
         idim_proxy_api = IdimProxyService(copy.deepcopy(self.requester_idir))