diff --git a/cloud/auth.py b/cloud/auth.py index 0b90cdd..48c50a6 100644 --- a/cloud/auth.py +++ b/cloud/auth.py @@ -334,7 +334,13 @@ class AuthStore: role = self._member_role(project_id, user_id) if role is None: return False - return _ROLE_LEVEL[role] >= _ROLE_LEVEL[min_role] + # 未知角色(如直接写库留下的 'member')一律按无权限处理, + # 绝不能 KeyError 冒泡成 500:权限检查的失败语义只能是 + # False / PermissionError,绝不是服务端异常。 + level = _ROLE_LEVEL.get(role) + if level is None: + return False + return level >= _ROLE_LEVEL[min_role] def require_permission(self, user_id, project_id, action): """不满足权限时 raise PermissionError。""" diff --git a/cloud/tests/test_auth.py b/cloud/tests/test_auth.py index 4be67b8..1557677 100644 --- a/cloud/tests/test_auth.py +++ b/cloud/tests/test_auth.py @@ -93,6 +93,31 @@ class TestAuth(unittest.TestCase): with self.assertRaises(PermissionError): self.store.require_permission(outsider_id, project_id, "read") + def test_unknown_role_denied_not_500(self): + # 回归:project_members 里直接写库留下的未知角色(如 + # 'member',set_member 本身会拦,但旧行/直写库拦不住) + # 在生产上曾让 check_permission 以 KeyError 冒泡成 500。 + # 权限检查的失败语义只能是 False / PermissionError。 + project_id, _o, viewer_id, _c, _m, _outsider = ( + self._project_with_roles() + ) + db_path = os.path.join(self.tmp.name, "auth.db") + conn = sqlite3.connect(db_path) + try: + conn.execute( + "UPDATE project_members SET role = ?" + " WHERE project_id = ? AND user_id = ?", + ("member", project_id, viewer_id)) + conn.commit() + finally: + conn.close() + self.assertFalse( + self.store.check_permission(viewer_id, project_id, "read")) + self.assertFalse( + self.store.check_permission(viewer_id, project_id, "create_task")) + with self.assertRaises(PermissionError): + self.store.require_permission(viewer_id, project_id, "read") + def test_permission_grants(self): project_id, owner_id, viewer_id, contributor_id, maintainer_id, _ = ( self._project_with_roles()