chakcy
f62704ca0e
一个插件 = 一个完整的 .svfs 包：客户端把插件前端经 svfs://<插件id>/ 直接加载 进 WebView（不落盘、不起端口、不解包），每个插件一个独立 origin（浏览器按 origin 隔离 localStorage）与独立键值存储；快照即版本——升级 = 派生快照的增量 打包，回滚 = 把注册表指针切回旧快照，包文件一个字节不用动。svfs_client 复用 svfs_studio 的主题与基础部件，故两个包一并入库。 按评估结论补齐的 P0 三块： * 会话：桥在"宿主注入的 token 被 401 拒绝"时通知壳层，提示一次并给出一键 「重新登录…」；服务器状态灯把「在线但登录失效」（黄）与「离线」（红）分开， 不再把凭据失效误报成网络不可达； * 凭据：新增 credentials.py——token 不再明文进 settings.json。Windows 走 DPAPI （密文只有当前用户在本机解得开），其它平台走系统凭据库（可选依赖 keyring）， 都没有才明文兜底并在「设置 → 登录凭据」如实提示；令牌与签发它的 base_url 绑定（换服务器不会把旧令牌发出去）；老字段 auth_token 启动时自动迁移； * 强登录链路保持：启动动画 → 静默校验 → 登录窗 → 主窗口，不可达时离线宽限。 P1/P2：net 超时与重试（只对幂等方法，避免重发 POST 变两单）可配、代理与 自签 CA 可配（loopback 永远直连）；诊断日志滚动落盘（1MB×3，可关）——崩溃后 还能回看现场；客户端自更新检查（静态 <base_url>/client/manifest.json，缺失 即静默）；插件清单可选 net_hosts 白名单；notify 改走系统托盘（无托盘降级状态栏）； 设置界面新增网络 / 登录凭据 / 诊断日志三块。 examples/：示例服务端（FastAPI，含登录、上传即 publish、客户端版本接口）、 Vue 3 插件模板（带浏览器版宿主模拟器）、svfs-plugins-root/ 为模板配套的演示 插件实体。测试：tests/test_client_*.py（纯层 + Qt 层 + 一次真实 WebEngine 冒烟）、 tests/test_studio_*.py、tests/test_plugin_server.py。
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()
|