Merge pull request #153 from openqj/feat-auth-unknown-role
fix(auth): unknown project role denies instead of KeyError 500
This commit is contained in:
2 files changed
+32
-1
No files matched your search
+7
-1
@@ -334,7 +334,13 @@ class AuthStore:
|
|||||||
role = self._member_role(project_id, user_id)
|
role = self._member_role(project_id, user_id)
|
||||||
if role is None:
|
if role is None:
|
||||||
return False
|
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):
|
def require_permission(self, user_id, project_id, action):
|
||||||
"""不满足权限时 raise PermissionError。"""
|
"""不满足权限时 raise PermissionError。"""
|
||||||
|
|||||||
@@ -93,6 +93,31 @@ class TestAuth(unittest.TestCase):
|
|||||||
with self.assertRaises(PermissionError):
|
with self.assertRaises(PermissionError):
|
||||||
self.store.require_permission(outsider_id, project_id, "read")
|
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):
|
def test_permission_grants(self):
|
||||||
project_id, owner_id, viewer_id, contributor_id, maintainer_id, _ = (
|
project_id, owner_id, viewer_id, contributor_id, maintainer_id, _ = (
|
||||||
self._project_with_roles()
|
self._project_with_roles()
|
||||||
|
|||||||
Reference in new issue
Block a user