host
c823e32336
check_permission now returns False for roles not in _ROLE_LEVEL (e.g. a directly-written 'member' row, which set_member itself rejects but legacy/direct DB rows can still carry), so the failure mode is PermissionError/403, never an unhandled 500. Regression test test_unknown_role_denied_not_500 fails with KeyError on old code.
175 lines
7.6 KiB
Python
175 lines
7.6 KiB
Python
"""auth 模块单测：API key 认证、角色权限、审计只增不减。"""
|
||
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.auth import ROLES, AuthStore
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.auth import ROLES, AuthStore
|
||
|
||
|
||
class TestAuth(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.store = AuthStore(os.path.join(self.tmp.name, "auth.db"))
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _user(self, username):
|
||
return self.store.create_user(username)
|
||
|
||
def _project_with_roles(self):
|
||
"""owner + viewer + contributor + maintainer + 一个非成员，返回其 id。"""
|
||
owner_id, _ = self._user("owner")
|
||
viewer_id, _ = self._user("viewer")
|
||
contributor_id, _ = self._user("contributor")
|
||
maintainer_id, _ = self._user("maintainer")
|
||
outsider_id, _ = self._user("outsider")
|
||
project_id = "proj-1"
|
||
self.store.create_project(project_id, owner_id)
|
||
self.store.set_member(project_id, owner_id, viewer_id, "viewer")
|
||
self.store.set_member(project_id, owner_id, contributor_id, "contributor")
|
||
self.store.set_member(project_id, owner_id, maintainer_id, "maintainer")
|
||
return project_id, owner_id, viewer_id, contributor_id, maintainer_id, outsider_id
|
||
|
||
def test_create_user_plain_key_authenticates(self):
|
||
user_id, api_key_plain = self._user("alice")
|
||
self.assertEqual(self.store.authenticate(api_key_plain), user_id)
|
||
|
||
def test_wrong_key_rejected(self):
|
||
user_id, api_key_plain = self._user("alice")
|
||
self.assertIsNone(self.store.authenticate(api_key_plain + "x"))
|
||
self.assertIsNone(self.store.authenticate("not-a-real-key"))
|
||
self.assertIsNone(self.store.authenticate(""))
|
||
self.assertIsNone(self.store.authenticate(None))
|
||
|
||
def test_hash_stored_not_plaintext(self):
|
||
_user_id, api_key_plain = self._user("alice")
|
||
db_path = os.path.join(self.tmp.name, "auth.db")
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT api_key_hash FROM users WHERE username = ?", ("alice",)
|
||
).fetchone()
|
||
finally:
|
||
conn.close()
|
||
self.assertIsNotNone(row)
|
||
stored = row[0]
|
||
self.assertNotEqual(stored, api_key_plain)
|
||
self.assertNotIn(api_key_plain, stored)
|
||
# 库里只有哈希列，没有明文列可查
|
||
columns = [c[1] for c in sqlite3.connect(db_path).execute(
|
||
"PRAGMA table_info(users)").fetchall()]
|
||
self.assertNotIn("api_key_plain", columns)
|
||
self.assertNotIn("api_key", columns)
|
||
|
||
def test_permission_denials(self):
|
||
project_id, _o, viewer_id, contributor_id, _m, outsider_id = (
|
||
self._project_with_roles()
|
||
)
|
||
# viewer 不能 create_task
|
||
self.assertFalse(
|
||
self.store.check_permission(viewer_id, project_id, "create_task"))
|
||
with self.assertRaises(PermissionError):
|
||
self.store.require_permission(viewer_id, project_id, "create_task")
|
||
# contributor 不能 approve_merge
|
||
self.assertFalse(
|
||
self.store.check_permission(contributor_id, project_id, "approve_merge"))
|
||
with self.assertRaises(PermissionError):
|
||
self.store.require_permission(
|
||
contributor_id, project_id, "approve_merge")
|
||
# 非成员读不到
|
||
self.assertFalse(
|
||
self.store.check_permission(outsider_id, project_id, "read"))
|
||
self.assertFalse(
|
||
self.store.check_permission(outsider_id, project_id, "create_task"))
|
||
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()
|
||
)
|
||
self.assertTrue(self.store.check_permission(viewer_id, project_id, "read"))
|
||
self.assertTrue(
|
||
self.store.check_permission(contributor_id, project_id, "create_task"))
|
||
self.assertTrue(
|
||
self.store.check_permission(contributor_id, project_id, "read"))
|
||
self.assertTrue(
|
||
self.store.check_permission(maintainer_id, project_id, "approve_merge"))
|
||
self.assertTrue(
|
||
self.store.check_permission(owner_id, project_id, "manage_project"))
|
||
self.assertFalse(
|
||
self.store.check_permission(maintainer_id, project_id, "manage_project"))
|
||
self.assertEqual(
|
||
ROLES, ["viewer", "contributor", "maintainer", "owner"])
|
||
|
||
def test_owner_can_set_member_maintainer_cannot(self):
|
||
project_id, owner_id, _v, _c, maintainer_id, _o = (
|
||
self._project_with_roles()
|
||
)
|
||
target_id, _ = self._user("target")
|
||
# owner 改成员成功
|
||
self.store.set_member(project_id, owner_id, target_id, "viewer")
|
||
self.assertTrue(self.store.check_permission(target_id, project_id, "read"))
|
||
self.assertFalse(
|
||
self.store.check_permission(target_id, project_id, "create_task"))
|
||
# maintainer 改成员被拒，且原角色不变
|
||
with self.assertRaises(PermissionError):
|
||
self.store.set_member(
|
||
project_id, maintainer_id, target_id, "contributor")
|
||
self.assertFalse(
|
||
self.store.check_permission(target_id, project_id, "create_task"))
|
||
|
||
def test_audit_append_only(self):
|
||
project_id, owner_id, _v, _c, _m, _o = self._project_with_roles()
|
||
self.assertFalse(hasattr(AuthStore, "delete_audit"))
|
||
self.assertFalse(hasattr(AuthStore, "update_audit"))
|
||
before = len(self.store.list_audit(project_id))
|
||
self.store.audit(owner_id, project_id, "create_task", "task-1")
|
||
self.store.audit(owner_id, project_id, "approve_merge", "task-1")
|
||
entries = self.store.list_audit(project_id)
|
||
self.assertEqual(len(entries), before + 2)
|
||
# 最新两条在前（倒序），且均为本次追加的动作
|
||
actions = [e["action"] for e in entries[:2]]
|
||
self.assertEqual(actions, ["approve_merge", "create_task"])
|
||
# 审计条目为 dict 且含关键字段
|
||
self.assertIn("user_id", entries[0])
|
||
self.assertIn("created_at", entries[0])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|