fix: avoid login tenant context import shadowing

This commit is contained in:
GuoQing Zhang
2026-07-17 10:39:51 +08:00
parent d268e8e8a2
commit 49261d5d63
2 changed files with 66 additions and 10 deletions
@@ -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():
@@ -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")