diff --git a/src/backend/bisheng/user/domain/services/user.py b/src/backend/bisheng/user/domain/services/user.py index 382262c2e..856f3569c 100644 --- a/src/backend/bisheng/user/domain/services/user.py +++ b/src/backend/bisheng/user/domain/services/user.py @@ -387,8 +387,6 @@ class UserService: # qualifies for login. Without bypass, every DAO below trips # NoTenantContextError because do_orm_execute can't infer a tenant for # tenant-aware tables (userrole/department/usergroup/roleaccess). - from bisheng.core.context.tenant import bypass_tenant_filter - with bypass_tenant_filter(): if role_ids is None: roles = await UserRoleDao.aget_user_roles(db_user.user_id) @@ -479,7 +477,10 @@ class UserService: NoTenantsAvailableError, TenantDisabledError, ) - from bisheng.core.context.tenant import DEFAULT_TENANT_ID, bypass_tenant_filter + from bisheng.database.models.department import ( + UserDepartmentDao, + ) + from bisheng.database.models.tenant import TenantDao, UserTenantDao # Reject login when the user's primary department mounts to a # disabled child tenant. TenantResolver intentionally walks past @@ -488,11 +489,6 @@ class UserService: # treats it as a member-level freeze. Without this guard a member # of a disabled tenant would get fallback-routed to Root and log # in normally, which contradicts the operator's intent. - from bisheng.database.models.department import ( - UserDepartmentDao, - ) - from bisheng.database.models.tenant import TenantDao, UserTenantDao - with bypass_tenant_filter(): primary_dept = await UserDepartmentDao.aget_user_primary_department( db_user.user_id, @@ -571,7 +567,6 @@ class UserService: # DB.status is authoritative; Redis blacklist is a defensive cross-check. if settings.multi_tenant.enabled and tenant_id and tenant_id > 0: from bisheng.common.errcode.tenant import TenantDisabledError - from bisheng.core.context.tenant import bypass_tenant_filter from bisheng.database.models.tenant import TenantDao with bypass_tenant_filter(): @@ -656,7 +651,6 @@ class UserService: extra_fields["tenant_id"] = tenant_id tenant_info = next((t for t in (tenants_list or []) if t.get("tenant_id") == tenant_id), None) if not tenant_info and settings.multi_tenant.enabled: - from bisheng.core.context.tenant import bypass_tenant_filter from bisheng.database.models.tenant import TenantDao with bypass_tenant_filter(): diff --git a/src/backend/test/user/test_user_access_request_dedup.py b/src/backend/test/user/test_user_access_request_dedup.py index 69a6ccdd2..3a66c5deb 100644 --- a/src/backend/test/user/test_user_access_request_dedup.py +++ b/src/backend/test/user/test_user_access_request_dedup.py @@ -105,3 +105,65 @@ async def test_login_access_guard_reuses_resolved_roles_and_department_flag(): role_ids=[2], is_department_admin=False, ) + + +async def test_user_login_enters_tenant_bypass_before_multi_tenant_resolution(): + db_user = SimpleNamespace(user_id=12, delete=0) + rejected_response = object() + login_request = SimpleNamespace(user_name="alice", password="encrypted") + captcha_setting = AsyncMock(return_value=False) + + with ( + patch.object( + user_service_module, + "settings", + SimpleNamespace(aget_from_db=captcha_setting), + ), + patch.object( + user_service_module.UserDao, + "aget_login_candidates_by_account", + new_callable=AsyncMock, + return_value=[db_user], + ), + patch.object( + user_service_module.UserService, + "decrypt_md5_password", + return_value="password", + ), + patch.object( + user_service_module.UserService, + "judge_user_password", + new_callable=AsyncMock, + ), + patch.object( + user_service_module.UserService, + "clear_error_password_key", + new_callable=AsyncMock, + ), + patch.object( + user_service_module.UserRoleDao, + "aget_user_roles", + new_callable=AsyncMock, + return_value=[], + ), + patch.object( + user_service_module.DepartmentDao, + "aget_user_admin_departments", + new_callable=AsyncMock, + return_value=[], + ), + patch.object( + user_service_module.UserService, + "_reject_login_if_user_has_no_usable_access", + new_callable=AsyncMock, + return_value=rejected_response, + ), + ): + result = await user_service_module.UserService.user_login( + SimpleNamespace(), + login_request, + auth_jwt=SimpleNamespace(), + ) + + assert result is rejected_response + captcha_setting.assert_awaited_once_with("use_captcha")