150 lines
6.4 KiB
Python
150 lines
6.4 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_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()
|