218 lines
9.2 KiB
Python
218 lines
9.2 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""登录凭据的安全存储测试（纯标准库层）。
|
|||
|
|
|
|||
|
|
重点是三件事：
|
|||
|
|
|
|||
|
|
1. **不再明文落盘**：settings.json 里不出现 token，凭据文件里也不是原文；
|
|||
|
|
2. **强度如实汇报**：DPAPI → 受保护；两个加密后端都不可用 → ``plain``
|
|||
|
|
且 ``is_secure()`` 为 False（界面据此提示）；
|
|||
|
|
3. **换服务器不串凭据**：令牌只交还给签发它的那个 base_url。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import base64
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import unittest
|
|||
|
|
|
|||
|
|
from helpers import VFSTestCase
|
|||
|
|
|
|||
|
|
from svfs_client import credentials, paths as client_paths, settings
|
|||
|
|
|
|||
|
|
assert sys.modules.get("helpers") is not None
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _FakeKeyring:
|
|||
|
|
"""keyring 的最小替身（够 credentials 用：后端名 + 三件套）。"""
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self.store = {}
|
|||
|
|
|
|||
|
|
def get_keyring(self):
|
|||
|
|
return "fake-keyring"
|
|||
|
|
|
|||
|
|
def set_password(self, service, username, password):
|
|||
|
|
self.store[(service, username)] = password
|
|||
|
|
|
|||
|
|
def get_password(self, service, username):
|
|||
|
|
return self.store.get((service, username))
|
|||
|
|
|
|||
|
|
def delete_password(self, service, username):
|
|||
|
|
self.store.pop((service, username), None)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class CredentialsTest(VFSTestCase):
|
|||
|
|
BASE_URL = "http://10.0.0.5:8000"
|
|||
|
|
|
|||
|
|
def setUp(self):
|
|||
|
|
super().setUp()
|
|||
|
|
self.base = client_paths.ensure_layout(self.path("data"))
|
|||
|
|
|
|||
|
|
# ---- 通用（任何后端都要满足） ------------------------------------
|
|||
|
|
|
|||
|
|
def test_roundtrip_and_base_url_gate(self):
|
|||
|
|
credentials.save(self.base, "tok-1", base_url=self.BASE_URL,
|
|||
|
|
username="admin")
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL), "tok-1")
|
|||
|
|
# 换服务器：旧令牌不交还（它只对签发它的服务器有效）
|
|||
|
|
self.assertEqual(credentials.load(self.base, "http://other:9000"), "")
|
|||
|
|
# 不指定服务器时不设门槛（调用方明确表示"就是当前这个"）
|
|||
|
|
self.assertEqual(credentials.load(self.base), "tok-1")
|
|||
|
|
|
|||
|
|
def test_token_never_in_settings(self):
|
|||
|
|
settings.save(self.base, {**settings.DEFAULTS,
|
|||
|
|
"base_url": self.BASE_URL,
|
|||
|
|
"auth_user": "admin"})
|
|||
|
|
credentials.save(self.base, "tok-secret", base_url=self.BASE_URL)
|
|||
|
|
raw = open(client_paths.settings_file(self.base), encoding="utf-8").read()
|
|||
|
|
self.assertNotIn("tok-secret", raw)
|
|||
|
|
|
|||
|
|
def test_empty_token_clears(self):
|
|||
|
|
credentials.save(self.base, "tok-1", base_url=self.BASE_URL)
|
|||
|
|
credentials.save(self.base, "", base_url=self.BASE_URL)
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL), "")
|
|||
|
|
self.assertFalse(os.path.exists(client_paths.credentials_file(self.base)))
|
|||
|
|
|
|||
|
|
def test_clear_removes_credential_file(self):
|
|||
|
|
credentials.save(self.base, "tok-1", base_url=self.BASE_URL)
|
|||
|
|
self.assertTrue(os.path.exists(client_paths.credentials_file(self.base)))
|
|||
|
|
credentials.clear(self.base)
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL), "")
|
|||
|
|
self.assertFalse(os.path.exists(client_paths.credentials_file(self.base)))
|
|||
|
|
|
|||
|
|
def test_corrupt_file_reads_as_logged_out(self):
|
|||
|
|
credentials.save(self.base, "tok-1", base_url=self.BASE_URL)
|
|||
|
|
with open(client_paths.credentials_file(self.base), "w",
|
|||
|
|
encoding="utf-8") as handle:
|
|||
|
|
handle.write("{ 不是 JSON")
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL), "")
|
|||
|
|
|
|||
|
|
def test_migrate_moves_settings_token(self):
|
|||
|
|
settings.save(self.base, {**settings.DEFAULTS, "base_url": self.BASE_URL,
|
|||
|
|
"auth_token": "tok-legacy",
|
|||
|
|
"auth_user": "admin"})
|
|||
|
|
self.assertEqual(credentials.migrate(self.base),
|
|||
|
|
credentials.backend(self.base))
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL),
|
|||
|
|
"tok-legacy")
|
|||
|
|
# 原字段清空（落盘的内容里不再有 token）
|
|||
|
|
self.assertEqual(settings.load(self.base)["auth_token"], "")
|
|||
|
|
raw = open(client_paths.settings_file(self.base), encoding="utf-8").read()
|
|||
|
|
self.assertNotIn("tok-legacy", raw)
|
|||
|
|
# 幂等：再迁一次什么也不做
|
|||
|
|
self.assertEqual(credentials.migrate(self.base), "")
|
|||
|
|
|
|||
|
|
def test_migrate_without_legacy_token_is_noop(self):
|
|||
|
|
settings.save(self.base, {**settings.DEFAULTS, "base_url": self.BASE_URL})
|
|||
|
|
self.assertEqual(credentials.migrate(self.base), "")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class PlainBackendTest(CredentialsTest):
|
|||
|
|
"""两个加密后端都不可用时的兜底路径（仍是原来的通用断言）。"""
|
|||
|
|
|
|||
|
|
def setUp(self):
|
|||
|
|
super().setUp()
|
|||
|
|
# 关掉 DPAPI 与 keyring：只剩 plain
|
|||
|
|
self._patch("_dpapi_usable", lambda: False)
|
|||
|
|
self._patch("_keyring_usable", lambda: False)
|
|||
|
|
self._patch("_keyring", lambda: None)
|
|||
|
|
|
|||
|
|
def _patch(self, name, value):
|
|||
|
|
original = getattr(credentials, name)
|
|||
|
|
setattr(credentials, name, value)
|
|||
|
|
self.addCleanup(setattr, credentials, name, original)
|
|||
|
|
|
|||
|
|
def test_reports_insecure(self):
|
|||
|
|
self.assertEqual(credentials.backend(self.base), "plain")
|
|||
|
|
self.assertFalse(credentials.is_secure(self.base))
|
|||
|
|
self.assertIn("明文", credentials.describe(self.base))
|
|||
|
|
|
|||
|
|
def test_encryption_failure_degrades_instead_of_losing_token(self):
|
|||
|
|
"""加密后端当场失败（DPAPI 被策略禁、keyring 后端挂）时降级明文，
|
|||
|
|
但不能把用户刚登录拿到的令牌弄丢。"""
|
|||
|
|
def boom(data, protect):
|
|||
|
|
raise OSError("DPAPI 被组策略禁用")
|
|||
|
|
|
|||
|
|
self._patch("_dpapi_usable", lambda: True)
|
|||
|
|
self._patch("_dpapi_call", boom)
|
|||
|
|
self.assertEqual(
|
|||
|
|
credentials.save(self.base, "tok-x", base_url=self.BASE_URL),
|
|||
|
|
"plain")
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL), "tok-x")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class KeyringBackendTest(CredentialsTest):
|
|||
|
|
"""keyring 可用（非 Windows 的常规路径）——文件里不该有令牌。"""
|
|||
|
|
|
|||
|
|
def setUp(self):
|
|||
|
|
super().setUp()
|
|||
|
|
self.fake = _FakeKeyring()
|
|||
|
|
self._patch("_dpapi_usable", lambda: False)
|
|||
|
|
self._patch("_keyring", lambda: self.fake)
|
|||
|
|
self._patch("_keyring_usable", lambda: True)
|
|||
|
|
|
|||
|
|
def _patch(self, name, value):
|
|||
|
|
original = getattr(credentials, name)
|
|||
|
|
setattr(credentials, name, value)
|
|||
|
|
self.addCleanup(setattr, credentials, name, original)
|
|||
|
|
|
|||
|
|
def test_token_lives_in_keyring_not_in_file(self):
|
|||
|
|
self.assertEqual(credentials.backend(self.base), "keyring")
|
|||
|
|
self.assertTrue(credentials.is_secure(self.base))
|
|||
|
|
credentials.save(self.base, "tok-keyring", base_url=self.BASE_URL)
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL),
|
|||
|
|
"tok-keyring")
|
|||
|
|
raw = open(client_paths.credentials_file(self.base),
|
|||
|
|
encoding="utf-8").read()
|
|||
|
|
self.assertNotIn("tok-keyring", raw)
|
|||
|
|
self.assertEqual(
|
|||
|
|
self.fake.store[("svfs-client", self.BASE_URL)], "tok-keyring")
|
|||
|
|
|
|||
|
|
def test_load_is_empty_when_keyring_backend_disappears(self):
|
|||
|
|
credentials.save(self.base, "tok-keyring", base_url=self.BASE_URL)
|
|||
|
|
self._patch("_keyring", lambda: None)
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL), "")
|
|||
|
|
|
|||
|
|
def test_clear_removes_keyring_entry(self):
|
|||
|
|
credentials.save(self.base, "tok-keyring", base_url=self.BASE_URL)
|
|||
|
|
credentials.clear(self.base)
|
|||
|
|
self.assertEqual(self.fake.store, {})
|
|||
|
|
|
|||
|
|
|
|||
|
|
@unittest.skipUnless(os.name == "nt", "DPAPI 只在 Windows 上存在")
|
|||
|
|
class DpapiBackendTest(CredentialsTest):
|
|||
|
|
"""Windows 默认后端：文件里是密文，且只有本机本用户能解。"""
|
|||
|
|
|
|||
|
|
def test_default_backend_is_dpapi(self):
|
|||
|
|
self.assertEqual(credentials.backend(self.base), "dpapi")
|
|||
|
|
self.assertTrue(credentials.is_secure(self.base))
|
|||
|
|
self.assertIn("DPAPI", credentials.describe(self.base))
|
|||
|
|
|
|||
|
|
def test_stored_bytes_are_ciphertext(self):
|
|||
|
|
credentials.save(self.base, "tok-secret-value", base_url=self.BASE_URL)
|
|||
|
|
with open(client_paths.credentials_file(self.base), "rb") as handle:
|
|||
|
|
raw = handle.read()
|
|||
|
|
self.assertNotIn(b"tok-secret-value", raw)
|
|||
|
|
record = json.loads(raw.decode("utf-8"))
|
|||
|
|
self.assertEqual(record["backend"], "dpapi")
|
|||
|
|
self.assertTrue(record["token"])
|
|||
|
|
# 存的是密文的 base64，不是令牌本身
|
|||
|
|
self.assertNotEqual(base64.b64decode(record["token"]),
|
|||
|
|
b"tok-secret-value")
|
|||
|
|
|
|||
|
|
def test_tampered_ciphertext_reads_as_logged_out(self):
|
|||
|
|
credentials.save(self.base, "tok-secret-value", base_url=self.BASE_URL)
|
|||
|
|
file = client_paths.credentials_file(self.base)
|
|||
|
|
record = json.load(open(file, encoding="utf-8"))
|
|||
|
|
blob = bytearray(base64.b64decode(record["token"]))
|
|||
|
|
blob[len(blob) // 2] ^= 0xFF # 篡改一个字节
|
|||
|
|
record["token"] = base64.b64encode(bytes(blob)).decode("ascii")
|
|||
|
|
with open(file, "w", encoding="utf-8") as handle:
|
|||
|
|
json.dump(record, handle)
|
|||
|
|
# 解不开就当没登录（不能崩，也不能返回垃圾）
|
|||
|
|
self.assertEqual(credentials.load(self.base, self.BASE_URL), "")
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
unittest.main()
|