host
5db89d1b06
Pre-fix any first-attempt audit failure raised NameError at auth.py:424 (sqlite3 never imported), aborting the 5-attempt retry loop on attempt 1 and masking the original error. The 424-line coverage mystery is also explained: the only full-suite hit is test_child_model.test_coordinator_inherits_child_defaults via worker best-effort _audit swallowing an OperationalError (database is locked); 422-423/426-427 stay uncovered because rollback succeeds and NameError aborts before either raise. Bubble audit (see report): no merge/release gate swallows the NameError and proceeds, so this stays a downgrade, not a production blocker. (cherry picked from commit 039c6adbac3b120b5b9a1dcbafeff6a91407ce70)
223 lines
9.6 KiB
Python
223 lines
9.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_retries_integrity_error_and_reraises_others(self):
|
||
"""A4 回归：audit() 重试分支的 sqlite3 引用必须存在。
|
||
|
||
首调 IntegrityError 后重试成功：审计对恰好一行（audit_log
|
||
1 行 + audit_ledger 1 行），不是 0 行也不是多行；非完整
|
||
性错误原样抛回（RuntimeError，不是 NameError 掩盖原错）。
|
||
背景：auth.py 曾全文件无 import sqlite3，audit() 的
|
||
except 分支在 424 行求 isinstance(exc, sqlite3.IntegrityError)
|
||
时直接 NameError，任何首调失败都走不到重试或原错抛回。
|
||
"""
|
||
import cloud.auth as auth_mod
|
||
|
||
original = auth_mod.audit_on_conn
|
||
calls = {"n": 0}
|
||
|
||
def _flaky(conn, *args, **kwargs):
|
||
calls["n"] += 1
|
||
if calls["n"] == 1:
|
||
raise sqlite3.IntegrityError(
|
||
"UNIQUE constraint failed: audit_ledger.entry_hash")
|
||
return original(conn, *args, **kwargs)
|
||
|
||
auth_mod.audit_on_conn = _flaky
|
||
try:
|
||
self.store.audit("u1", "p1", "act", "d")
|
||
finally:
|
||
auth_mod.audit_on_conn = original
|
||
self.assertEqual(calls["n"], 2)
|
||
self.assertEqual(len(self.store.list_audit("p1")), 1)
|
||
ledger_rows = self.store.conn.execute(
|
||
"SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
|
||
self.assertEqual(ledger_rows, 1)
|
||
|
||
def _broken(conn, *args, **kwargs):
|
||
raise RuntimeError("ledger boom")
|
||
|
||
auth_mod.audit_on_conn = _broken
|
||
try:
|
||
with self.assertRaises(RuntimeError) as caught:
|
||
self.store.audit("u1", "p1", "act", "d2")
|
||
finally:
|
||
auth_mod.audit_on_conn = original
|
||
self.assertEqual(str(caught.exception), "ledger boom")
|
||
self.assertEqual(len(self.store.list_audit("p1")), 1)
|
||
ledger_rows = self.store.conn.execute(
|
||
"SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
|
||
self.assertEqual(ledger_rows, 1)
|
||
|
||
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()
|