325 lines
12 KiB
Python
325 lines
12 KiB
Python
|
"""完整性：每文件 sha256、快照级摘要、HMAC 签名。
|
||
|
|
|
||
|
|
设计上的一个重要区分：
|
||
|
|
* ``verify()`` 是**元数据级**校验（摘要、blob 是否存在、大小是否一致），很便宜；
|
||
|
|
* ``verify(check_content=True)`` 才会逐文件重算 sha256，是**权威**校验，
|
||
|
|
代价是 O(总字节数)。同大小的内容篡改只有后者能发现。
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import sqlite3
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
import helpers
|
||
|
|
|
||
|
|
from sqlite_vfs.core import SQLiteVFS, VFSIntegrityError
|
||
|
|
|
||
|
|
|
||
|
|
class DigestTest(helpers.VFSTestCase):
|
||
|
|
def _pack(self, name="d.svfs", tree=None):
|
||
|
|
src = helpers.make_tree(
|
||
|
|
self.path(f"src_{name}"),
|
||
|
|
tree or {"a.txt": b"aaa", "sub/b.txt": b"bbb"},
|
||
|
|
)
|
||
|
|
db = self.path(name)
|
||
|
|
helpers.pack(src, db, compress=True, name="digest-pack")
|
||
|
|
return db
|
||
|
|
|
||
|
|
def test_digest_is_recorded_and_stable_across_reopen(self):
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
first = vfs.get_stats()["digest"]
|
||
|
|
self.assertTrue(first)
|
||
|
|
vfs.close()
|
||
|
|
|
||
|
|
reopened = SQLiteVFS(db)
|
||
|
|
self.addCleanup(reopened.close)
|
||
|
|
self.assertEqual(first, reopened.get_stats()["digest"])
|
||
|
|
|
||
|
|
def test_digest_matches_manual_recomputation(self):
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
stored = vfs.conn.execute("SELECT digest FROM snapshots WHERE id = 1").fetchone()[0]
|
||
|
|
self.assertEqual(stored, vfs.compute_digest(1, store=False))
|
||
|
|
|
||
|
|
def test_digest_changes_when_a_file_is_added(self):
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
before = vfs.compute_digest(1)
|
||
|
|
|
||
|
|
extra = helpers.write_file(self.path("extra.txt"), b"extra")
|
||
|
|
with helpers.quiet():
|
||
|
|
vfs.add_file(extra, "extra.txt")
|
||
|
|
vfs.recount()
|
||
|
|
self.assertNotEqual(before, vfs.compute_digest(1))
|
||
|
|
|
||
|
|
def test_digest_changes_when_content_changes(self):
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
before = vfs.compute_digest(1)
|
||
|
|
|
||
|
|
vfs.remove("a.txt")
|
||
|
|
changed = helpers.write_file(self.path("changed.txt"), b"zzz")
|
||
|
|
with helpers.quiet():
|
||
|
|
vfs.add_file(changed, "a.txt")
|
||
|
|
vfs.recount()
|
||
|
|
self.assertNotEqual(before, vfs.compute_digest(1))
|
||
|
|
|
||
|
|
def test_digest_changes_when_metadata_changes(self):
|
||
|
|
"""路径/大小/时间戳/权限也参与摘要，不只是内容。"""
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
before = vfs.compute_digest(1)
|
||
|
|
|
||
|
|
vfs.conn.execute("UPDATE files SET permissions = 511 WHERE path = 'a.txt'")
|
||
|
|
vfs.conn.commit()
|
||
|
|
self.assertNotEqual(before, vfs.compute_digest(1))
|
||
|
|
|
||
|
|
def test_digest_ignores_insertion_order(self):
|
||
|
|
"""摘要必须只取决于内容与元数据，与插入顺序无关。
|
||
|
|
|
||
|
|
用 seed_rows 造数据（时间戳为 NULL），这样两个库的元数据完全一致，
|
||
|
|
唯一的差别就是插入顺序。
|
||
|
|
"""
|
||
|
|
rows = [
|
||
|
|
{"path": "a.txt", "content": b"a"},
|
||
|
|
{"path": "b.txt", "content": b"b"},
|
||
|
|
{"path": "c.txt", "content": b"c"},
|
||
|
|
]
|
||
|
|
digests = []
|
||
|
|
for index, ordered in enumerate((list(rows), list(reversed(rows)))):
|
||
|
|
vfs = SQLiteVFS(self.path(f"order{index}.svfs"))
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
helpers.seed_rows(vfs, ordered)
|
||
|
|
digests.append(vfs.compute_digest(1))
|
||
|
|
self.assertEqual(digests[0], digests[1])
|
||
|
|
|
||
|
|
|
||
|
|
class VerifyTest(helpers.VFSTestCase):
|
||
|
|
def _pack(self, name="v.svfs", compress=True):
|
||
|
|
src = helpers.make_tree(
|
||
|
|
self.path(f"src_{name}"),
|
||
|
|
{"a.txt": b"aaa", "big.txt": b"repeat me " * 300, "sub/b.txt": b"bbb"},
|
||
|
|
)
|
||
|
|
db = self.path(name)
|
||
|
|
helpers.pack(src, db, compress=compress, name="verify-pack")
|
||
|
|
return db
|
||
|
|
|
||
|
|
def test_verify_passes_on_a_fresh_pack(self):
|
||
|
|
vfs = SQLiteVFS(self._pack())
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
report = vfs.verify()
|
||
|
|
self.assertTrue(report["ok"], report["errors"])
|
||
|
|
self.assertTrue(report["digest_ok"])
|
||
|
|
self.assertEqual(3, report["files_checked"])
|
||
|
|
self.assertEqual(0, report["content_checked"], "默认不做逐文件哈希")
|
||
|
|
|
||
|
|
def test_verify_with_content_check_passes(self):
|
||
|
|
vfs = SQLiteVFS(self._pack())
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
report = vfs.verify(check_content=True)
|
||
|
|
self.assertTrue(report["ok"], report["errors"])
|
||
|
|
self.assertEqual(3, report["content_checked"])
|
||
|
|
|
||
|
|
def test_verify_reports_empty_database(self):
|
||
|
|
vfs = SQLiteVFS(":memory:")
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
report = vfs.verify()
|
||
|
|
self.assertFalse(report["ok"])
|
||
|
|
self.assertIn("没有", " ".join(report["errors"]))
|
||
|
|
|
||
|
|
def test_content_tamper_is_caught_by_check_content(self):
|
||
|
|
"""同大小改掉内容：元数据级校验看不出来，权威校验必须抓到。"""
|
||
|
|
db = self._pack(compress=False)
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
|
||
|
|
vfs.conn.execute("UPDATE blobs SET content = X'626262' WHERE size = 3")
|
||
|
|
vfs.conn.commit()
|
||
|
|
|
||
|
|
# 元数据级：摘要与大小都没变，所以看不出来
|
||
|
|
self.assertTrue(vfs.verify()["ok"])
|
||
|
|
|
||
|
|
report = vfs.verify(check_content=True)
|
||
|
|
self.assertFalse(report["ok"])
|
||
|
|
self.assertTrue(any("内容哈希不匹配" in e for e in report["errors"]), report)
|
||
|
|
|
||
|
|
def test_read_file_with_verify_hash_raises_on_tamper(self):
|
||
|
|
db = self._pack(compress=False)
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.conn.execute("UPDATE blobs SET content = X'626262' WHERE size = 3")
|
||
|
|
vfs.conn.commit()
|
||
|
|
|
||
|
|
self.assertEqual(b"bbb", vfs.read_file("a.txt")) # 不校验时照旧返回
|
||
|
|
with self.assertRaises(VFSIntegrityError):
|
||
|
|
vfs.read_file("a.txt", verify_hash=True)
|
||
|
|
|
||
|
|
def test_missing_blob_is_caught(self):
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.conn.execute("DELETE FROM blobs WHERE hash = (SELECT content_hash FROM files WHERE path='a.txt')")
|
||
|
|
vfs.conn.commit()
|
||
|
|
|
||
|
|
report = vfs.verify()
|
||
|
|
self.assertFalse(report["ok"])
|
||
|
|
self.assertTrue(any("blob 缺失" in e for e in report["errors"]), report)
|
||
|
|
with self.assertRaises(VFSIntegrityError):
|
||
|
|
vfs.read_file("a.txt")
|
||
|
|
|
||
|
|
def test_size_mismatch_is_caught(self):
|
||
|
|
db = self._pack(compress=False)
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.conn.execute("UPDATE blobs SET size = size + 1")
|
||
|
|
vfs.conn.commit()
|
||
|
|
|
||
|
|
report = vfs.verify()
|
||
|
|
self.assertFalse(report["ok"])
|
||
|
|
self.assertTrue(any("大小不一致" in e for e in report["errors"]), report)
|
||
|
|
with self.assertRaises(VFSIntegrityError):
|
||
|
|
vfs.read_file("a.txt")
|
||
|
|
|
||
|
|
def test_metadata_tamper_is_caught_by_digest(self):
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.conn.execute("UPDATE files SET file_size = 999 WHERE path = 'a.txt'")
|
||
|
|
vfs.conn.commit()
|
||
|
|
|
||
|
|
report = vfs.verify()
|
||
|
|
self.assertFalse(report["ok"])
|
||
|
|
self.assertFalse(report["digest_ok"])
|
||
|
|
self.assertTrue(any("摘要不一致" in e for e in report["errors"]), report)
|
||
|
|
|
||
|
|
def test_path_tamper_is_caught_by_digest(self):
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.conn.execute("UPDATE files SET path = 'renamed.txt' WHERE path = 'a.txt'")
|
||
|
|
vfs.conn.commit()
|
||
|
|
self.assertFalse(vfs.verify()["digest_ok"])
|
||
|
|
|
||
|
|
def test_verify_can_target_an_older_snapshot(self):
|
||
|
|
"""摘要防的是"绕过 API 的改动"；走 API 的写入会刷新摘要，不算篡改。"""
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.create_snapshot("v2", copy_entries=True)
|
||
|
|
vfs.remove("a.txt")
|
||
|
|
vfs.recount()
|
||
|
|
|
||
|
|
# 走 API 删文件之后 recount 会刷新摘要，所以两个快照此刻都是自洽的
|
||
|
|
self.assertTrue(vfs.verify(snapshot_id=2)["ok"])
|
||
|
|
self.assertTrue(vfs.verify(snapshot_id=1)["ok"])
|
||
|
|
|
||
|
|
# 绕过 API 直接改 v2 的元数据 → 只有 v2 报错
|
||
|
|
vfs.conn.execute("UPDATE files SET file_size = 1 WHERE snapshot_id = 2")
|
||
|
|
vfs.conn.commit()
|
||
|
|
self.assertFalse(vfs.verify(snapshot_id=2)["digest_ok"])
|
||
|
|
self.assertTrue(
|
||
|
|
vfs.verify(snapshot_id=1)["ok"], "v1 没被碰过，不该受影响"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class SignatureTest(helpers.VFSTestCase):
|
||
|
|
def _pack(self, name="sig.svfs"):
|
||
|
|
src = helpers.make_tree(self.path("src"), {"a.txt": b"aaa", "b.txt": b"bbb"})
|
||
|
|
db = self.path(name)
|
||
|
|
helpers.pack(src, db, compress=True, name="signed-pack")
|
||
|
|
return db
|
||
|
|
|
||
|
|
def test_sign_and_verify(self):
|
||
|
|
vfs = SQLiteVFS(self._pack())
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
signature = vfs.sign("top-secret", key_id="release-1")
|
||
|
|
self.assertEqual(64, len(signature), "sha256 十六进制应当是 64 个字符")
|
||
|
|
|
||
|
|
report = vfs.verify_signature("top-secret")
|
||
|
|
self.assertTrue(report["ok"], report["errors"])
|
||
|
|
self.assertEqual("release-1", report["key_id"])
|
||
|
|
self.assertEqual("hmac-sha256", vfs.list_signatures()[0]["algorithm"])
|
||
|
|
|
||
|
|
def test_wrong_key_fails(self):
|
||
|
|
vfs = SQLiteVFS(self._pack())
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.sign("right-key")
|
||
|
|
report = vfs.verify_signature("wrong-key")
|
||
|
|
self.assertFalse(report["ok"])
|
||
|
|
self.assertTrue(report["errors"])
|
||
|
|
|
||
|
|
def test_tampering_after_signing_fails(self):
|
||
|
|
vfs = SQLiteVFS(self._pack())
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.sign("k")
|
||
|
|
|
||
|
|
extra = helpers.write_file(self.path("evil.txt"), b"evil")
|
||
|
|
with helpers.quiet():
|
||
|
|
vfs.add_file(extra, "evil.txt")
|
||
|
|
vfs.recount()
|
||
|
|
|
||
|
|
self.assertFalse(
|
||
|
|
vfs.verify_signature("k")["ok"], "签名之后内容变了，验签必须失败"
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_verify_without_a_signature_is_reported(self):
|
||
|
|
vfs = SQLiteVFS(self._pack())
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
report = vfs.verify_signature("k")
|
||
|
|
self.assertFalse(report["ok"])
|
||
|
|
self.assertIn("没有签名记录", " ".join(report["errors"]))
|
||
|
|
|
||
|
|
def test_verify_signature_by_key_id(self):
|
||
|
|
vfs = SQLiteVFS(self._pack())
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
vfs.sign("k1", key_id="one")
|
||
|
|
vfs.sign("k2", key_id="two")
|
||
|
|
self.assertTrue(vfs.verify_signature("k1", key_id="one")["ok"])
|
||
|
|
self.assertTrue(vfs.verify_signature("k2", key_id="two")["ok"])
|
||
|
|
self.assertFalse(vfs.verify_signature("k1", key_id="two")["ok"])
|
||
|
|
|
||
|
|
def test_signature_verifies_on_a_readonly_connection(self):
|
||
|
|
"""只读方必须能在不改动库的前提下完成验证。"""
|
||
|
|
db = self._pack()
|
||
|
|
vfs = SQLiteVFS(db)
|
||
|
|
vfs.sign("shared-secret", key_id="audit")
|
||
|
|
vfs.close()
|
||
|
|
|
||
|
|
before = helpers.md5_of(db)
|
||
|
|
readonly = SQLiteVFS(db, readonly=True)
|
||
|
|
self.addCleanup(readonly.close)
|
||
|
|
|
||
|
|
report = readonly.verify_signature("shared-secret")
|
||
|
|
self.assertTrue(report["ok"], report["errors"])
|
||
|
|
self.assertTrue(readonly.verify(check_content=True)["ok"])
|
||
|
|
self.assertEqual(before, helpers.md5_of(db), "只读验证改动了库文件")
|
||
|
|
|
||
|
|
def test_sign_requires_write_access(self):
|
||
|
|
db = self._pack()
|
||
|
|
readonly = SQLiteVFS(db, readonly=True)
|
||
|
|
self.addCleanup(readonly.close)
|
||
|
|
with self.assertRaises(PermissionError):
|
||
|
|
readonly.sign("k")
|
||
|
|
|
||
|
|
def test_digest_can_be_exported_for_external_signing(self):
|
||
|
|
"""非对称签名要交给外部工具，所以摘要必须能干净地取出来。"""
|
||
|
|
vfs = SQLiteVFS(self._pack())
|
||
|
|
self.addCleanup(vfs.close)
|
||
|
|
payload = {
|
||
|
|
"digest": vfs.compute_digest(1, store=False),
|
||
|
|
"snapshot_id": 1,
|
||
|
|
"algorithm": "sha256",
|
||
|
|
}
|
||
|
|
text = json.dumps(payload, sort_keys=True)
|
||
|
|
self.assertIn("digest", text)
|
||
|
|
self.assertEqual(64, len(payload["digest"]))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|