242 lines
9.4 KiB
Python
Executable File
242 lines
9.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""内部狗粮审核账号与 RBAC 安全边界测试。"""
|
|
|
|
from pathlib import Path
|
|
import importlib.util
|
|
import os
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
TOOL = ROOT / "deploy" / "manage-dogfood-reviewers.py"
|
|
STAGING_DB = ROOT / "game-cloud" / "huijing-server" / "src" / "main" / "resources" / "db" / "staging"
|
|
STAGING_CONFIG = ROOT / "game-cloud" / "huijing-server" / "src" / "main" / "resources" / "application-staging.yaml"
|
|
|
|
|
|
class ReviewerSecurityTest(unittest.TestCase):
|
|
|
|
@staticmethod
|
|
def load_module():
|
|
spec = importlib.util.spec_from_file_location("dogfood_reviewers", TOOL)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
def test_staging_explicitly_disables_mock_authentication(self):
|
|
config = STAGING_CONFIG.read_text(encoding="utf-8")
|
|
self.assertIn("mock-enable: false", config)
|
|
self.assertNotIn("mock-enable: true", config)
|
|
|
|
def test_rbac_seed_is_repeatable_and_contains_no_login_credentials(self):
|
|
self.assertFalse((STAGING_DB / "V34.0.0__seed_dogfood_review_rbac.sql").exists())
|
|
migration = STAGING_DB / "R__seed_dogfood_review_rbac.sql"
|
|
sql = migration.read_text(encoding="utf-8")
|
|
self.assertNotIn("INSERT INTO system_users", sql)
|
|
self.assertNotIn("$2a$", sql)
|
|
self.assertIn("dogfood_reviewer", sql)
|
|
self.assertIn("dogfood_readonly", sql)
|
|
|
|
def test_rotation_uses_exact_roles_and_invalidates_old_tokens(self):
|
|
module = self.load_module()
|
|
api = FakeApi(module)
|
|
|
|
result = module.rotate_reviewers(api, module.generate_password)
|
|
|
|
self.assertEqual({"dogfood_reviewer", "dogfood_readonly", "admin"}, set(result["accounts"]))
|
|
self.assertEqual([201], api.assigned_roles[101])
|
|
self.assertEqual([202], api.assigned_roles[102])
|
|
self.assertEqual([1, 0], api.status_updates[101])
|
|
self.assertEqual([1, 0], api.status_updates[102])
|
|
self.assertEqual([1], api.status_updates[1])
|
|
self.assertEqual(("status", 1, 1), api.events[-2])
|
|
self.assertEqual(("login-rejected", "admin", "admin123"), api.events[-1])
|
|
self.assertNotEqual(result["accounts"]["dogfood_reviewer"],
|
|
result["accounts"]["dogfood_readonly"])
|
|
for password in result["accounts"].values():
|
|
self.assertEqual(16, len(password))
|
|
self.assertNotEqual("admin123", password)
|
|
self.assertTrue(any(char.islower() for char in password))
|
|
self.assertTrue(any(char.isupper() for char in password))
|
|
self.assertTrue(any(char.isdigit() for char in password))
|
|
|
|
def test_rotation_creates_missing_accounts_before_fail_closed_status_cycle(self):
|
|
module = self.load_module()
|
|
api = FakeApi(module, existing=False)
|
|
|
|
result = module.rotate_reviewers(api, module.generate_password)
|
|
|
|
self.assertEqual(["dogfood_reviewer", "dogfood_readonly"], api.created_users)
|
|
self.assertEqual([1], api.updated_password_users)
|
|
self.assertEqual([1, 0], api.status_updates[101])
|
|
self.assertEqual([1, 0], api.status_updates[102])
|
|
self.assertEqual([1], api.status_updates[1])
|
|
self.assertEqual(3, len(result["evidenceAccounts"]))
|
|
|
|
def test_api_rbac_verification_reads_full_menu_permissions(self):
|
|
module = self.load_module()
|
|
responses = FakeRbacResponses()
|
|
api = module.ApiClient("test-admin-token")
|
|
api._request = responses.request
|
|
|
|
roles = api.verify_rbac()
|
|
|
|
self.assertEqual(set(module.REVIEWER_PERMISSIONS), roles["dogfood_reviewer"]["permissions"])
|
|
self.assertIn(("GET", "/system/menu/list"), responses.calls)
|
|
self.assertNotIn(("GET", "/system/menu/simple-list"), responses.calls)
|
|
evidence = module.build_rbac_evidence(roles)
|
|
self.assertEqual(["dogfood_readonly", "dogfood_reviewer"],
|
|
[item["roleCode"] for item in evidence])
|
|
|
|
def test_sensitive_files_require_private_permissions_and_no_overwrite(self):
|
|
module = self.load_module()
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
token_file = root / "admin.token"
|
|
token_file.write_text("secret-token\n", encoding="utf-8")
|
|
os.chmod(token_file, 0o600)
|
|
self.assertEqual("secret-token", module.require_private_file(token_file))
|
|
|
|
output = root / "credentials.json"
|
|
module.reserve_private_file(output)
|
|
self.assertEqual(0o600, stat.S_IMODE(output.stat().st_mode))
|
|
with self.assertRaises(FileExistsError):
|
|
module.reserve_private_file(output)
|
|
|
|
os.chmod(token_file, 0o644)
|
|
with self.assertRaises(module.ReviewerProvisionError):
|
|
module.require_private_file(token_file)
|
|
|
|
symlink = root / "admin-token-link"
|
|
symlink.symlink_to(token_file)
|
|
with self.assertRaises(module.ReviewerProvisionError):
|
|
module.require_private_file(symlink)
|
|
|
|
def test_default_admin_negative_login_is_anonymous_and_fail_closed(self):
|
|
module = self.load_module()
|
|
captured_requests = []
|
|
original_urlopen = module.urlopen
|
|
|
|
def rejected_urlopen(request, timeout):
|
|
captured_requests.append(request)
|
|
return FakeHttpResponse({"code": 1_002_000_001, "msg": "账号被禁用", "data": None})
|
|
|
|
module.urlopen = rejected_urlopen
|
|
try:
|
|
api = module.ApiClient("bootstrap-token")
|
|
self.assertEqual(1_002_000_001,
|
|
api.verify_password_login_rejected("admin", "admin123"))
|
|
self.assertIsNone(captured_requests[0].get_header("Authorization"))
|
|
|
|
module.urlopen = lambda request, timeout: FakeHttpResponse({
|
|
"code": 0,
|
|
"msg": "",
|
|
"data": {"accessToken": "unexpected"},
|
|
})
|
|
with self.assertRaises(module.ReviewerProvisionError):
|
|
api.verify_password_login_rejected("admin", "admin123")
|
|
finally:
|
|
module.urlopen = original_urlopen
|
|
|
|
|
|
class FakeApi:
|
|
"""仅模拟账号轮换所需接口，断言调用结果而非 HTTP 实现细节。"""
|
|
|
|
def __init__(self, module, existing=True):
|
|
self.module = module
|
|
self.existing = existing
|
|
self.assigned_roles = {}
|
|
self.status_updates = {1: [], 101: [], 102: []}
|
|
self.created_users = []
|
|
self.updated_password_users = []
|
|
self.events = []
|
|
|
|
def verify_rbac(self):
|
|
return {
|
|
"dogfood_reviewer": {"id": 201, "permissions": set(self.module.REVIEWER_PERMISSIONS)},
|
|
"dogfood_readonly": {"id": 202, "permissions": set(self.module.READONLY_PERMISSIONS)},
|
|
}
|
|
|
|
def find_user(self, username):
|
|
if username == "admin":
|
|
return {"id": 1, "username": username}
|
|
if not self.existing:
|
|
return None
|
|
return {"id": 101 if username == "dogfood_reviewer" else 102, "username": username}
|
|
|
|
def create_user(self, username, nickname, password):
|
|
if self.existing:
|
|
raise AssertionError("测试账号已存在，不应走创建分支")
|
|
self.assert_password(password)
|
|
self.created_users.append(username)
|
|
return 101 if username == "dogfood_reviewer" else 102
|
|
|
|
def update_password(self, user_id, password):
|
|
self.assert_password(password)
|
|
self.updated_password_users.append(user_id)
|
|
self.events.append(("password", user_id))
|
|
|
|
def assign_user_roles(self, user_id, role_ids):
|
|
self.assigned_roles[user_id] = role_ids
|
|
|
|
def update_user_status(self, user_id, status):
|
|
self.status_updates[user_id].append(status)
|
|
self.events.append(("status", user_id, status))
|
|
|
|
def get_user_role_ids(self, user_id):
|
|
return set(self.assigned_roles[user_id])
|
|
|
|
def verify_password_login_rejected(self, username, password):
|
|
self.events.append(("login-rejected", username, password))
|
|
return 1_002_000_000
|
|
|
|
@staticmethod
|
|
def assert_password(password):
|
|
if len(password) != 16:
|
|
raise AssertionError("随机密码长度不符合约束")
|
|
|
|
|
|
class FakeRbacResponses:
|
|
"""模拟管理 API 响应，校验工具使用真实包含 permission 的菜单入口。"""
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def request(self, method, path, **kwargs):
|
|
self.calls.append((method, path))
|
|
if path == "/system/role/simple-list":
|
|
return [
|
|
{"id": 201, "code": "dogfood_reviewer"},
|
|
{"id": 202, "code": "dogfood_readonly"},
|
|
]
|
|
if path == "/system/menu/list":
|
|
return [
|
|
{"id": 301, "permission": "project:review:query"},
|
|
{"id": 302, "permission": "project:review:approve"},
|
|
{"id": 303, "permission": "telemetry:game-stat:query"},
|
|
]
|
|
role_id = int(kwargs["query"]["roleId"])
|
|
return [301, 302, 303] if role_id == 201 else [301, 303]
|
|
|
|
|
|
class FakeHttpResponse:
|
|
"""提供 urllib 响应所需的最小上下文管理协议。"""
|
|
|
|
def __init__(self, payload):
|
|
import json
|
|
self.body = json.dumps(payload).encode("utf-8")
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
return False
|
|
|
|
def read(self):
|
|
return self.body
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|