fix(incubator): 身份判定限定「昆明市大学生创业园」本园

本应用仅服务昆明市大学生创业园(按名称精确定位 park_tenants):
- 管理方 = 本园管理员(operator_user_id 或 admin_username 绑定,当前 admin_username=yunpeng)
- 提交方 = 本园(tenant_id=T001)的在园企业成员
- 其他账号 → 无权限,明确提示'不属于昆明市大学生创业园'
(移除此前 operator/carrier 任意角色放宽;数据范围本已按 identity.tenant_id 隔离)
This commit is contained in:
Pine
2026-09-01 21:16:26 +08:00
parent 04624ab477
commit 684f4253b9
2 changed files with 33 additions and 31 deletions
+1 -1
View File
@@ -115,7 +115,7 @@ async def incubator_me(
ident = await resolve_identity(db, user)
if not ident.is_manager and not ident.company_id:
return {"identity": None,
"message": "当前账号未绑定园区入驻企业,或非园区运营方,请联系运营方绑定企业"}
"message": "当前账号不属于昆明市大学生创业园:非本园入驻企业成员,也非本园管理员,请联系园区管理员绑定"}
return {"identity": ident.to_dict()}
+32 -30
View File
@@ -31,7 +31,9 @@ from ..infrastructure.models import (
)
from ..infrastructure.repositories import utcnow_iso
# 管理权限码(与 seed.py role=carrier 权限对齐)
# 本应用唯一服务的园区(昆明市大学生创业园):身份与数据范围都限定该园区。
# 管理权限码(与 seed.py role=carrier 权限对齐,保留供细粒度授权扩展)
TARGET_PARK_NAME = "昆明市大学生创业园"
MANAGE_PERMS = ("menu:park_incubator", "incubator:manage")
@@ -61,55 +63,55 @@ class Identity:
async def resolve_identity(db, user: dict) -> Identity:
"""解析当前用户在创业园应用中的生效身份(管理方 / 提交方 / 无权限)
"""解析当前用户在「昆明市大学生创业园应用中的生效身份。
返回 Identity;无权限时 is_manager=False 且 company_id 为空(调用方按 403 处理)。
本应用仅服务「昆明市大学生创业园」这一个园区(按名称精确定位,
不依赖园区 id,环境里若有同名园区即为目标):
- 管理方 = 该园区的管理员(park_tenants.operator_user_id 或 admin_username 绑定)
- 提交方 = 该园区(tenant_id 一致)的在园企业成员
- 其他 = 无权限(不属于本园区)
"""
uid = user.get("id", "")
username = user.get("username", "")
role = user.get("role", "")
sub_role = user.get("sub_role", "")
perms = set(user.get("permissions") or [])
# —— 管理方:carrier 归属园区 / 具备管理权限码 ——
if role == "carrier" or perms & set(MANAGE_PERMS):
trow = (await db.session.execute(
select(ParkTenant).where(ParkTenant.operator_user_id == uid,
ParkTenant.status == "active")
)).scalar_one_or_none()
if trow is not None:
return Identity(is_manager=True, tenant_id=trow.id, company_id="",
company_name="", role=f"carrier:{sub_role or 'admin'}",
user_id=uid)
# carrier 但未绑定园区 → 允许其选择园区(取第一个 active 园区,便于管理端绑定)
first = (await db.session.execute(
select(ParkTenant).where(ParkTenant.status == "active").limit(1)
)).scalar_one_or_none()
if first is not None:
return Identity(is_manager=True, tenant_id=first.id, company_id="",
company_name="", role="carrier", user_id=uid)
# 目标园区:昆明市大学生创业园(active)
target = (await db.session.execute(
select(ParkTenant).where(ParkTenant.name == TARGET_PARK_NAME,
ParkTenant.status == "active")
)).scalar_one_or_none()
if target is None:
# 环境未配置目标园区 → 一律无权限
return Identity(is_manager=False, tenant_id="", company_id="",
company_name="", role=role or "opc_member", user_id=uid)
tid = target.id
# —— 提交方:园区内入驻企业成员 ——
# —— 管理方:目标园区的管理员(operator_user_id 或 admin_username 绑定) ——
if target.operator_user_id == uid or target.admin_username == username:
return Identity(is_manager=True, tenant_id=tid, company_id="",
company_name="", role="carrier:admin", user_id=uid)
# —— 提交方:目标园区的在园企业成员 ——
rows = (await db.session.execute(
select(CompanyMember).where(CompanyMember.user_id == uid,
CompanyMember.status == "active")
)).scalars().all()
for m in rows:
comp = await db.session.get(ParkCompany, m.company_id)
if comp is None or not comp.tenant_id:
if comp is None or comp.tenant_id != tid:
continue
return Identity(is_manager=False, tenant_id=comp.tenant_id,
company_id=comp.id, company_name=comp.name,
role="opc_member", user_id=uid)
return Identity(is_manager=False, tenant_id=tid, company_id=comp.id,
company_name=comp.name, role="opc_member", user_id=uid)
# —— 无权限 ——
# —— 无权限(非本园区账号) ——
return Identity(is_manager=False, tenant_id="", company_id="",
company_name="", role=role or "opc_member", user_id=uid)
def require_identity(identity: Identity) -> None:
"""身份不可用(非管理且未绑定企业)时抛异常。"""
"""身份不可用(非昆明市大学生创业园管理且未绑定本园企业)时抛异常。"""
if not identity.is_manager and not identity.company_id:
raise LookupError("当前账号未绑定园区入驻企业,或非园区运营方,无法使用本应用")
raise LookupError("当前账号不属于昆明市大学生创业园,无法使用本应用")
# ===========================================================================