From 25431b42f4e3bd6392e5b27c88724d0b947935b3 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Wed, 30 Sep 2026 07:04:31 +0800 Subject: [PATCH 1/5] Verify installed-package concurrency, crash recovery and full restore conservation Refs #65 #69 #80 #83 #84 #85. Measure bounded disposable workload; preserve every row and sequence across real SIGKILL and backup restore. Production evidence remains separate. --- PROGRESS.md | 59 +++++++++ docs/RELEASE_ACCEPTANCE.md | 15 ++- scripts/check_installed_server.py | 199 ++++++++++++++++++++++++++++-- 3 files changed, 260 insertions(+), 13 deletions(-) create mode 100644 PROGRESS.md diff --git a/PROGRESS.md b/PROGRESS.md new file mode 100644 index 00000000..f0950857 --- /dev/null +++ b/PROGRESS.md @@ -0,0 +1,59 @@ +# 交付进度 · 2026-09-30 + +目标：逐项处理全部开放 issue/PR，交付经验证的安装包和部署步骤。 +进度以实际提交、测试运行和现场材料为准，不把旧记录或测试数量当完成率。 + +## 本轮基线与完成事项 + +- 已核对 16 个开放 issue（#64、#65、#68、#69、#70、#72、#76–#85）与唯一开放 PR #186。 +- 已读取当前权威设计及有效归档；现有功能和历史协议保持原义。 +- #186 已审查并合并为 `dfb89fe989eaef66c99f14ae6a953b07204b7170`。 + 主线 tree `e1d65b624266527bef5f66d2843fa1a40c4ddc5f` 与实际通过测试的 tree 相同。 +- [核心验收 36632579472](https://github.com/TokenNotIncluded/msg.lmm.best/actions/runs/36632579472)： + 八份 ZIP 摘要、manifest 和 JUnit 真实 node-ID 已核验，2541 core + 8 conformance， + 无失败、错误、跳过、重复或遗漏。 +- [质量 36632578929](https://github.com/TokenNotIncluded/msg.lmm.best/actions/runs/36632578929)： + Ruff JSON 零诊断、stderr 为空、585 文件格式、Python 3.15.0rc2 编译通过。 +- [部署演练 36632578807](https://github.com/TokenNotIncluded/msg.lmm.best/actions/runs/36632578807)： + 同一 wheel 的锁定 server/client 安装、真实启动/重启、四传输、只读 doctor、 + 禁外发 worker、真实 OpenSSH/Git 撤权及正式隔离 selftest 42 项均通过。 + wheel/sdist 与核心构建产物逐字节一致；精确摘要见 PR #186。 + +## 当前推进 + +- 增加安装包的有界并发写入/读取测量、真实 SIGKILL 重启与全表/sequence 不变量检查。 +- 增加安装包的真实 PostgreSQL/Git/CAS 备份恢复计时和全表值比较；只允许恢复流程 + 明确规定的 runtime_config/quarantine 变化，验证删 marker 后仍隔离业务与幂等重放。 +- 这些新检查尚待云端执行，不能使用 #186 的旧成功结果标记新检查通过。 +- 逐条核对 issue 中尚未刷新到当前源码的描述，保留原验收范围和历史证据。 + +## issue 交接 + +| issue | 已有代码/隔离证据入口 | 本轮需要继续完成的范围 | +| --- | --- | --- | +| #64 | URL/Host/真实隔离 Nginx 日志回归；#186 主体读 effect 栅栏 | 实际 listener/CDN/proxy/APM/日志链材料 | +| #65 | 固定旧源码 fixture、LedgerAccount 迁移/回滚 | 真实旧快照的合法来源、冻结点及快照演练 | +| #68 | 历史密文迁移、逐项 ACK、备份退役证明校验 | 真实保留密文及独立备份范围退役证据 | +| #69 | 完整 proof/current-policy reconcile/promotion、持久隔离 | rollback 集之外的真实当前 pin 与本机受控恢复 | +| #70 | Root/CA/控制台拒绝与 bank fund 源码回归 | 真实 VT/串口及有限存量 CA 审核 | +| #72 | Bounty/官方 market_e2e/邮件/恢复/版本政策回归 | 对照原清单逐断言验收，更新过时的 #103 描述 | +| #76 | 多来源授权及跨协议/恢复矩阵 | 完整投影、缓存、当前权限条款核对 | +| #77 | 私有内容、独立签名、Legacy、荣誉挑战与展示 | 全条款/客户端/当前失效事实核对 | +| #78 | ReadQuery/Search/Sync/游标及 #186 | 完整成本/表示/当前授权矩阵核对 | +| #79 | 文件、patch/rebase、Revision、原子发布/物理失败 | 完整内容/文件条款核对 | +| #80 | Transfer/Git/LFS/hosting/一致存储及真实 SSH | 共享卷/多实例目标拓扑容量与持久性 | +| #81 | 通知/EffectJob/协作/SMTP socket/Webhook 回归 | 全事件契约与受控部署接收端 | +| #82 | CLI/TUI/工具真实隔离/SSH/安装包 | 目标机路径/PAM/systemd 与全部客户端条款 | +| #83 | 配置、Registry、规则及有限设计分母 | 逐条断言映射、当前文档与真实现场闸门 | +| #84 | 锁定安装包、隔离服务/SSH/selftest 演练 | 本轮补测量；真实主机/旧数据/切流/回滚仍需材料 | +| #85 | 唯一市场 owner、受保护账本/版本兼容/严格质量 | 对当前热点/重复实现核对，不按旧目录诊断 | + +## 完成条件与外部阻塞 + +新改动须在同一最终 tree 上通过质量、完整八分片/协议、安装包演练与适用专项； +合并后再核对独立 main push。问题解决后才更新状态，不能删断言或把取消当通过。 + +现场尚缺：目标主机只读入口与配置/日志链、受保护真实旧备份及来源、真实本机 +Root/CA 操作者、独立当前检查点、密文/备份退役材料、目标容量/停止/回滚条件。 +所需材料及操作步骤见 `docs/RELEASE_ACCEPTANCE.md` 和 `docs/DEPLOYMENT.md`。 +当前没有生产部署、资金操作、真实外发、备份删除或恢复放行。 diff --git a/docs/RELEASE_ACCEPTANCE.md b/docs/RELEASE_ACCEPTANCE.md index 3aaf8da4..fac82afc 100644 --- a/docs/RELEASE_ACCEPTANCE.md +++ b/docs/RELEASE_ACCEPTANCE.md @@ -1,7 +1,8 @@ # 发布验收与现场交接 -本页是 PR #185 的当前入口，取代把历史进度数字当作当前部署结论的做法。 -源码基线为已合并 #176 的 `25cf518a`；#182/#183/#184 均已在该历史。 +本页是发布验收的入口，取代把历史进度数字当作当前部署结论的做法。 +本轮基线为已合并 #186 的 `dfb89fe9`，tree `e1d65b62`；#185 及其前置修复已在历史。 +本轮事项与未完成范围见仓库根目录 `PROGRESS.md`。 每次最终验证以产物 `source.json` 中的 commit/tree、对应 workflow 和完整 JUnit node-ID 为准。PR 的模拟 merge commit 与分支 head 可不同，必须核对实际 tree。 本页索引测试，不单凭测试文件名声称原设计所有分支或生产已经验收。 @@ -23,7 +24,15 @@ server 包在 checkout 外运行，Root 目录确实不可由网络账号读取 client-only 包不得安装或导入 server runtime；其传输检查使用 MockTransport， 不冒称真实远程服务器连接。安装包正式 selftest 的 Test Root 也不代表真实物理控制台。 -通过结果与 ZIP/wheel/锁文件摘要附在 PR #185 和 #83，不为写入成功数字再改代码树。 +通过结果与 ZIP/wheel/锁文件摘要附在对应 PR 和 #83，不为写入成功数字再改代码树。 + +安装包演练还执行固定四个独立签名客户端的 32 次写入与 32 次读取，输出实际 +耗时/吞吐/p50/p95、daemon RSS/峰值 RSS、内容文件字节和剩余磁盘。真实 SIGKILL +后重启须保留所有表值、sequence 和同请求幂等；真实 PostgreSQL/Git/CAS 备份 +恢复须逐表逐值比较，仅允许确切的 runtime_config 暂停及持久 quarantine 改动。 +删除 marker 后重新创建应用仍须隔离业务与幂等重放。失败直接使部署演练失败。 +这些数值仅代表该次有界 CI 工作负载，不是目标主机容量或负载上限，也不取代 +共享卷、多实例、断电、旧生产数据及真实恢复 pin 的现场验收。 ## 开放 issue 的源码与验收入口 diff --git a/scripts/check_installed_server.py b/scripts/check_installed_server.py index ffbaef1b..85b8d525 100644 --- a/scripts/check_installed_server.py +++ b/scripts/check_installed_server.py @@ -11,11 +11,15 @@ import hashlib import importlib.metadata import json +import math import os +import platform +import shutil import socket import subprocess import sys import tempfile +import time from pathlib import Path import httpx @@ -23,12 +27,13 @@ from psycopg import sql import msg +from msg.admin.backups import backup, restore from msg.admin.diagnostics import temporary_postgres from msg.admin.root import _approve_csr, _provision from msg.application import Application from msg.client import ClientState, MsgClient -from msg.config import write_example -from msg.core.codec import canonical +from msg.config import load_settings, write_example +from msg.core.codec import canonical, loads from msg.transports.client import ( GraphQLTransport, HTTPTransport, @@ -41,20 +46,17 @@ def run(*command): return subprocess.run(command, capture_output=True, text=True, timeout=60, check=True) -def database_digest(dsn): +def database_facts(dsn): """Read complete row/sequence values without advancing PostgreSQL sequences.""" - hasher = hashlib.sha256() + facts = {'tables': {}, 'sequences': {}} with psycopg.connect(dsn) as connection: connection.execute('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY') tables = connection.execute( "SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY tablename" ).fetchall() for (table,) in tables: - hasher.update(canonical(table)) rows = connection.execute(sql.SQL('SELECT * FROM {}').format(sql.Identifier(table))) - for row in sorted(canonical(row) for row in rows): - hasher.update(len(row).to_bytes(8, 'big')) - hasher.update(row) + facts['tables'][table] = sorted(canonical(row) for row in rows) sequences = connection.execute( "SELECT sequencename FROM pg_sequences WHERE schemaname='public' ORDER BY sequencename" ).fetchall() @@ -62,8 +64,145 @@ def database_digest(dsn): value = connection.execute( sql.SQL('SELECT last_value,is_called FROM {}').format(sql.Identifier(sequence)) ).fetchone() - hasher.update(canonical((sequence, value))) - return hasher.hexdigest() + facts['sequences'][sequence] = canonical(value) + return facts + + +def database_digest(dsn): + return hashlib.sha256(canonical(database_facts(dsn))).hexdigest() + + +def latency_report(samples, elapsed): + ordered = sorted(samples) + assert ordered and elapsed > 0 + return { + 'completed': len(ordered), + 'elapsed_seconds': elapsed, + 'operations_per_second': len(ordered) / elapsed, + 'latency_p50_seconds': ordered[math.ceil(len(ordered) * 0.5) - 1], + 'latency_p95_seconds': ordered[math.ceil(len(ordered) * 0.95) - 1], + 'latency_max_seconds': ordered[-1], + } + + +async def concurrent_workload(folder, origin, http, dsn): + """Bounded workload on independently signed clients, not production load.""" + writers = [] + for index in range(4): + client = MsgClient( + ClientState(folder / f'capacity-client-{index}', server=origin), + HTTPTransport(origin, http=http), + ) + assert (await client.register(f'capacity-client-{index}')).status == 'ok' + writers.append(client) + + async def write(client, index): + measured = [] + for offset in range(8): + body = f'capacity-{index}-{offset}\n' + 'bounded content\n' * 64 + packet = client.prepare( + 'content.post_create', + {'parent': '/main', 'body': body}, + request_id=f'capacity-{index}-{offset}', + ) + started = time.perf_counter() + posted = await client.send(packet) + measured.append((time.perf_counter() - started, posted.resources, body)) + assert posted.status == 'ok', 'Concurrent signed write failed' + assert len(posted.resources) == 1 + return measured + + started = time.perf_counter() + written = await asyncio.gather(*(write(client, i) for i, client in enumerate(writers))) + writes = latency_report( + [row[0] for group in written for row in group], time.perf_counter() - started + ) + assert len({row[1][0].id for group in written for row in group}) == 32 + before = database_digest(dsn) + + async def read(client, group): + measured = [] + for _, references, body in group: + started = time.perf_counter() + result = await client.call('discovery.get', {'id': references[0].id}) + measured.append(time.perf_counter() - started) + assert result.status == 'ok' and result.data['content'] == body + return measured + + started = time.perf_counter() + reads = await asyncio.gather( + *(read(client, group) for client, group in zip(writers, written, strict=True)) + ) + reading = latency_report( + [sample for group in reads for sample in group], time.perf_counter() - started + ) + assert database_digest(dsn) == before, 'Concurrent reads changed authoritative facts' + return {'concurrent_clients': 4, 'writes': writes, 'reads': reading} + + +async def restore_drill(settings, folder, client, packet, source_dsn): + """Compare every row and sequence, allowing only exact quarantine changes.""" + app = Application(settings) + try: + await app.load() + original = database_facts(source_dsn) + started = time.perf_counter() + archive = folder / 'recovery.zip' + result = await backup(app, archive) + backup_seconds = time.perf_counter() - started + assert result['root_private_key_included'] is False + assert database_facts(source_dsn) == original + finally: + await app.close() + with temporary_postgres() as destination_dsn: + started = time.perf_counter() + restored = restore( + archive, folder / 'restore-etc', folder / 'restore-data', postgres_dsn=destination_dsn + ) + restore_seconds = time.perf_counter() - started + assert restored['promotion'] == 'blocked' + current = database_facts(destination_dsn) + expected_settings = dict(loads(row) for row in original['tables']['settings']) + current_settings = dict(loads(row) for row in current['tables']['settings']) + quarantine = loads(current_settings['recovery_quarantine']) + assert quarantine == { + 'format': 'msg-recovery-quarantine-v1', + 'outbound_enabled': False, + 'source_backup_sha256': result['sha256'], + 'revocation_replay': 'required', + 'authority': 'health_only', + } + runtime = loads(expected_settings.get('runtime_config', '{}')) + expected_settings['runtime_config'] = canonical({ + **runtime, + 'accept_writes': False, + 'cleanup_enabled': False, + }).decode() + expected_settings['recovery_quarantine'] = current_settings['recovery_quarantine'] + original['tables']['settings'] = sorted(canonical(row) for row in expected_settings.items()) + assert current == original, 'Restore changed other table values or sequence state' + for marker_present in (True, False): + if not marker_present: + (folder / 'restore-etc/recovery-drill.json').unlink() + recovered = Application(load_settings(folder / 'restore-etc')) + try: + await recovered.load() + replay = await recovered.executor.execute(packet) + assert replay.error.code == 'writes_paused' + read = client.prepare('discovery.get', {'id': client.state.subject}) + denied = await recovered.executor.execute(read) + assert denied.error.code == 'recovery_quarantined' + assert database_facts(destination_dsn) == current + finally: + await recovered.close() + return { + 'backup_seconds': backup_seconds, + 'restore_seconds': restore_seconds, + 'archive_bytes': archive.stat().st_size, + 'all_table_values_and_sequences_preserved': True, + 'persistent_quarantine_without_marker': True, + 'promotion_performed': False, + } async def acceptance(): @@ -71,6 +210,7 @@ async def acceptance(): assert 'site-packages' in Path(msg.__file__).parts, 'Use the installed wheel with -I' assert msg.__version__ == importlib.metadata.version('msgctl') checks = ['installed_wheel_and_version'] + measurements = {'platform': platform.platform(), 'cpu_count': os.cpu_count()} with ( tempfile.TemporaryDirectory(prefix='msg-installed-rehearsal-') as temporary, temporary_postgres() as dsn, @@ -128,7 +268,9 @@ async def start(): child.wait(timeout=15) raise + started = time.perf_counter() process = await start() + measurements['startup_seconds'] = time.perf_counter() - started checks.append('real_daemon_startup') state = ClientState(folder / 'client', server=origin) async with httpx.AsyncClient(timeout=10, trust_env=False) as http: @@ -162,6 +304,27 @@ async def start(): assert read.status == 'ok' and read.data['content'] == 'installed release\r\n' assert database_digest(dsn) == before checks.append('four_real_transports_preserve_all_database_facts') + measurements['workload'] = await concurrent_workload(folder, origin, http, dsn) + checks.append('concurrent_signed_writes_and_read_conservation') + status = (Path('/proc') / str(process.pid) / 'status').read_text() + measurements['daemon_memory_kib'] = { + name: int(line.split()[1]) + for line in status.splitlines() + for name in ('VmRSS:', 'VmHWM:') + if line.startswith(name) + } + before = database_digest(dsn) + process.kill() + process.wait(timeout=15) + started = time.perf_counter() + process = await start() + measurements['sigkill_restart_seconds'] = time.perf_counter() - started + assert database_digest(dsn) == before + repeated = await client.send(packet) + assert repeated.status == 'ok' and repeated.replayed + assert repeated.resources == posted.resources + assert database_digest(dsn) == before + checks.append('sigkill_restart_preserves_all_facts_and_idempotency') before = database_digest(dsn) diagnosed = json.loads( run(str(executable), '--config-dir', str(settings.config_dir), 'doctor').stdout @@ -175,6 +338,20 @@ async def start(): process.wait(timeout=15) process = None checks.append('clean_daemon_shutdown') + measurements['recovery'] = await restore_drill(settings, folder, client, packet, dsn) + checks.append('installed_backup_restore_conservation_and_persistent_quarantine') + measurements['content_file_bytes'] = sum( + entry.stat().st_size + for directory in ( + settings.server.content_dir, + settings.server.repositories_dir, + settings.server.blob_dir, + settings.server.staging_dir, + ) + for entry in directory.rglob('*') + if entry.is_file() + ) + measurements['disk_free_bytes'] = shutil.disk_usage(folder).free finally: if process is not None and process.poll() is None: process.terminate() @@ -190,6 +367,8 @@ async def start(): 'version': msg.__version__, 'temporary_installation_removed': True, 'external_recipients': 0, + 'measurements': measurements, + 'measurement_scope': 'bounded disposable CI workload; not target-host capacity', } From 2a2a8ac72f4c29e8bbd04510df750e4291e52f1d Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Wed, 30 Sep 2026 07:13:40 +0800 Subject: [PATCH 2/5] Fix restoring packed private Git refs before content verification --- PROGRESS.md | 5 ++- src/msg/admin/backups.py | 66 ++++++++++--------------------------- tests/test_backup_drill.py | 67 +++++++++++++++++++++++++++++++++++++- 3 files changed, 88 insertions(+), 50 deletions(-) diff --git a/PROGRESS.md b/PROGRESS.md index f0950857..10aa1508 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -24,7 +24,10 @@ - 增加安装包的有界并发写入/读取测量、真实 SIGKILL 重启与全表/sequence 不变量检查。 - 增加安装包的真实 PostgreSQL/Git/CAS 备份恢复计时和全表值比较；只允许恢复流程 明确规定的 runtime_config/quarantine 变化，验证删 marker 后仍隔离业务与幂等重放。 -- 这些新检查尚待云端执行，不能使用 #186 的旧成功结果标记新检查通过。 +- 首次新演练 [36643217304](https://github.com/TokenNotIncluded/msg.lmm.best/actions/runs/36643217304) + 在真实恢复时失败：私有 Git 打包引用后，ZIP 丢失空 refs 目录，Git 无法识别仓库。 + 本轮修复私有/用户仓库统一结构目录重建，保持只读 proof 校验不修改磁盘， + 并新增 packed-ref 恢复、只读拒绝和符号链接拒绝回归。新最终 tree 仍须重新完整验收。 - 逐条核对 issue 中尚未刷新到当前源码的描述，保留原验收范围和历史证据。 ## issue 交接 diff --git a/src/msg/admin/backups.py b/src/msg/admin/backups.py index dc09abdf..86b3d81c 100644 --- a/src/msg/admin/backups.py +++ b/src/msg/admin/backups.py @@ -62,6 +62,19 @@ def _db_refs(connection): return {'contents': contents, 'revisions': revisions} +def _git_layout(repository, *, repair): + """Restore empty bare-Git directories without changing objects or refs.""" + require(repository.is_dir() and not repository.is_symlink(), 'backup_repository_invalid') + for name in ('objects', 'objects/info', 'objects/pack', 'refs', 'refs/heads', 'refs/tags'): + path = repository / name + require(not path.is_symlink(), 'backup_symlink_forbidden') + require(not path.exists() or path.is_dir(), 'backup_git_layout_invalid') + if repair: + path.mkdir(exist_ok=True) + else: + require(path.is_dir(), 'backup_git_layout_missing') + + def _verify_storage(directory, refs, *, settings=None, repair_git_layout=True): """Verify archived bytes, Git objects and the SQL-referenced content index.""" content = settings.server.content_dir if settings is not None else directory / 'content' @@ -74,6 +87,9 @@ def _verify_storage(directory, refs, *, settings=None, repair_git_layout=True): and blobs.is_dir(), 'backup_storage_missing', ) + # ZIP contains files only. Packed refs can leave the private repository's + # entire refs directory empty, which Git still requires to recognize it. + _git_layout(content / 'private.git', repair=repair_git_layout) index_files = {path.name: path for path in (content / 'index').iterdir() if path.is_file()} require(set(refs['contents']).issubset(index_files), 'backup_content_missing') for key, index in index_files.items(): @@ -129,13 +145,7 @@ def _verify_storage(directory, refs, *, settings=None, repair_git_layout=True): require( repository.is_dir() and repository.name.endswith('.git'), 'backup_repository_invalid' ) - # ZIP stores files only. An empty bare repository still needs these - # structural directories; recreating them changes no Git object/ref. - for name in ('objects/info', 'objects/pack', 'refs/heads', 'refs/tags'): - if repair_git_layout: - (repository / name).mkdir(parents=True, exist_ok=True) - else: - require((repository / name).is_dir(), 'backup_git_layout_missing') + _git_layout(repository, repair=repair_git_layout) process = subprocess.run( ['git', '--git-dir', str(repository), 'fsck', '--full', '--no-reflogs'], capture_output=True, @@ -411,47 +421,7 @@ def restore(source, config_dir, data_dir, *, postgres_dsn='service=msgd'): _db_refs(connection) == manifest['references'], 'backup_database_reference_mismatch' ) row = connection.execute( - "SELECT value FROM settings WHERE key='runtime_config'" - ).fetchone() - runtime = loads(row[0]) if row is not None else {} - require(type(runtime) is dict, 'backup_runtime_config_invalid') - runtime['accept_writes'] = False - runtime['cleanup_enabled'] = False - connection.execute( - """INSERT INTO settings(key,value) VALUES('runtime_config',%s) - ON CONFLICT(key) DO UPDATE SET value=EXCLUDED.value""", - (canonical(runtime).decode(),), - ) - data_dir.mkdir(mode=0o700) - settings = write_example( - config_dir, data_dir, manifest['service_url'], postgres_dsn=postgres_dsn - ) - # Install the fail-closed marker as soon as a runnable config exists, - # including if a later filesystem verification fails. - durable_write( - config_dir / 'recovery-drill.json', - canonical({**quarantine, 'format': 'msg-recovery-drill-v1'}), - mode=0o600, - ) - _write_restored_config(directory / 'source-config.toml', settings, postgres_dsn) - settings = load_settings(config_dir) - for name, target in ( - ('content', settings.server.content_dir), - ('repositories', settings.server.repositories_dir), - ('blobs', settings.server.blob_dir), - ('staging', settings.server.staging_dir), - ): - if (directory / name).exists(): - target.parent.mkdir(parents=True, exist_ok=True) - shutil.move(str(directory / name), target) - settings.service_keys.parent.mkdir(parents=True, exist_ok=True) - shutil.move(str(directory / 'service'), settings.service_keys) - os.chmod(settings.service_keys, 0o700) - settings.trust_file.parent.mkdir(parents=True, exist_ok=True) - from msg.security.trust_files import write_trust - - write_trust( - settings.config_dir, + "SELECT value FROM settings…492 tokens truncated… settings.config_dir, loads((directory / 'root-public.json').read_bytes()), writer=durable_write, ) diff --git a/tests/test_backup_drill.py b/tests/test_backup_drill.py index 6469ab3a..94b9ff7e 100644 --- a/tests/test_backup_drill.py +++ b/tests/test_backup_drill.py @@ -11,7 +11,7 @@ import pytest from test_service import NOW, call, register -from msg.admin.backups import backup, restore +from msg.admin.backups import _db_refs, _verify_storage, backup, restore from msg.application import Application from msg.config import load_settings from msg.core.codec import canonical, loads @@ -23,6 +23,71 @@ from msg.workers.maintenance import run_maintenance +@pytest.mark.asyncio +async def test_restore_private_git_after_all_refs_are_packed(installed, tmp_path, pg_dsn): + app, _ = installed + key, subject, _ = await register(app, 'backup-packed') + created = await call( + app, + 'content.post_create', + {'parent': '/main', 'body': 'preserve packed private content'}, + key=key, + subject=subject, + ) + assert created.status == 'ok' + repository = app.settings.server.content_dir / 'private.git' + app.contents._run('pack-refs', '--all', '--prune') + assert (repository / 'packed-refs').is_file() + assert not any(path.is_file() for path in (repository / 'refs').rglob('*')) + archive = tmp_path / 'packed.zip' + await backup(app, archive) + with zipfile.ZipFile(archive) as source: + assert not any(name.startswith('content/private.git/refs/') for name in source.namelist()) + references = loads(source.read('manifest.json'))['references'] + config, data = tmp_path / 'packed-etc', tmp_path / 'packed-data' + result = restore(archive, config, data, postgres_dsn=pg_dsn) + assert result['promotion'] == 'blocked' + settings = load_settings(config) + _verify_storage(data, references, settings=settings, repair_git_layout=False) + with psycopg.connect(pg_dsn) as connection: + assert _db_refs(connection) == references + assert connection.execute( + 'SELECT body FROM resources WHERE id=%s', (created.resources[0].id,) + ).fetchone() is not None + quarantine = connection.execute( + "SELECT value FROM settings WHERE key='recovery_quarantine'" + ).fetchone() + assert loads(quarantine[0])['authority'] == 'health_only' + + +@pytest.mark.asyncio +async def test_readonly_storage_verification_does_not_repair_private_git(installed, tmp_path): + app, _ = installed + directory = app.settings.server.content_dir / 'private.git' / 'refs' / 'tags' + directory.rmdir() + with pytest.raises(Failure, match='backup_git_layout_missing'): + _verify_storage( + tmp_path, + {'contents': {}, 'revisions': []}, + settings=app.settings, + repair_git_layout=False, + ) + assert not directory.exists() + + +@pytest.mark.asyncio +async def test_storage_layout_repair_rejects_symlink(installed, tmp_path): + app, _ = installed + outside = tmp_path / 'outside' + outside.mkdir() + directory = app.settings.server.content_dir / 'private.git' / 'refs' / 'tags' + directory.rmdir() + directory.symlink_to(outside, target_is_directory=True) + with pytest.raises(Failure, match='backup_symlink_forbidden'): + _verify_storage(tmp_path, {'contents': {}, 'revisions': []}, settings=app.settings) + assert not tuple(outside.iterdir()) + + @pytest.mark.asyncio async def test_pg_dump_and_file_copy_hold_writer_lock(installed, tmp_path, pg_dsn, monkeypatch): app, _ = installed From b5c833ab4bc7a0a9affb20612398dc42e01b02c8 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Wed, 30 Sep 2026 07:16:40 +0800 Subject: [PATCH 3/5] Preserve complete backup source and canonical regression formatting --- src/msg/admin/backups.py | 42 +++++++++++++++++++++++++++++++++++++- tests/test_backup_drill.py | 5 +++-- 2 files changed, 44 insertions(+), 3 deletions(-) diff --git a/src/msg/admin/backups.py b/src/msg/admin/backups.py index 86b3d81c..ee142413 100644 --- a/src/msg/admin/backups.py +++ b/src/msg/admin/backups.py @@ -421,7 +421,47 @@ def restore(source, config_dir, data_dir, *, postgres_dsn='service=msgd'): _db_refs(connection) == manifest['references'], 'backup_database_reference_mismatch' ) row = connection.execute( - "SELECT value FROM settings…492 tokens truncated… settings.config_dir, + "SELECT value FROM settings WHERE key='runtime_config'" + ).fetchone() + runtime = loads(row[0]) if row is not None else {} + require(type(runtime) is dict, 'backup_runtime_config_invalid') + runtime['accept_writes'] = False + runtime['cleanup_enabled'] = False + connection.execute( + """INSERT INTO settings(key,value) VALUES('runtime_config',%s) + ON CONFLICT(key) DO UPDATE SET value=EXCLUDED.value""", + (canonical(runtime).decode(),), + ) + data_dir.mkdir(mode=0o700) + settings = write_example( + config_dir, data_dir, manifest['service_url'], postgres_dsn=postgres_dsn + ) + # Install the fail-closed marker as soon as a runnable config exists, + # including if a later filesystem verification fails. + durable_write( + config_dir / 'recovery-drill.json', + canonical({**quarantine, 'format': 'msg-recovery-drill-v1'}), + mode=0o600, + ) + _write_restored_config(directory / 'source-config.toml', settings, postgres_dsn) + settings = load_settings(config_dir) + for name, target in ( + ('content', settings.server.content_dir), + ('repositories', settings.server.repositories_dir), + ('blobs', settings.server.blob_dir), + ('staging', settings.server.staging_dir), + ): + if (directory / name).exists(): + target.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(directory / name), target) + settings.service_keys.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(directory / 'service'), settings.service_keys) + os.chmod(settings.service_keys, 0o700) + settings.trust_file.parent.mkdir(parents=True, exist_ok=True) + from msg.security.trust_files import write_trust + + write_trust( + settings.config_dir, loads((directory / 'root-public.json').read_bytes()), writer=durable_write, ) diff --git a/tests/test_backup_drill.py b/tests/test_backup_drill.py index 94b9ff7e..a1f69fce 100644 --- a/tests/test_backup_drill.py +++ b/tests/test_backup_drill.py @@ -51,9 +51,10 @@ async def test_restore_private_git_after_all_refs_are_packed(installed, tmp_path _verify_storage(data, references, settings=settings, repair_git_layout=False) with psycopg.connect(pg_dsn) as connection: assert _db_refs(connection) == references - assert connection.execute( + resource = connection.execute( 'SELECT body FROM resources WHERE id=%s', (created.resources[0].id,) - ).fetchone() is not None + ).fetchone() + assert resource is not None quarantine = connection.execute( "SELECT value FROM settings WHERE key='recovery_quarantine'" ).fetchone() From 6d4843d1543122f7f9a05ae8b8fb733dfd9c8dd8 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Wed, 30 Sep 2026 07:20:15 +0800 Subject: [PATCH 4/5] Verify deferred bounty COMMIT and every honor context boundary --- tests/test_achievement_context_boundary.py | 121 +++++++++++++++++++++ tests/test_bounty_commit_boundary.py | 60 ++++++++++ 2 files changed, 181 insertions(+) create mode 100644 tests/test_achievement_context_boundary.py create mode 100644 tests/test_bounty_commit_boundary.py diff --git a/tests/test_achievement_context_boundary.py b/tests/test_achievement_context_boundary.py new file mode 100644 index 00000000..bbb8a63c --- /dev/null +++ b/tests/test_achievement_context_boundary.py @@ -0,0 +1,121 @@ +"""Every round is bound to its nonce/deadline/context; final issue is unique.""" + +import asyncio +from datetime import timedelta + +import pytest +from read_only_evidence import business_snapshot +from test_achievements import advance +from test_service import NOW, call, register + +from msg.core.codec import loads, wire +from msg.plugins.achievements import FINAL_STATEMENT + + +def confirmation(challenge): + result = { + name: challenge[name] + for name in ('challenge_id', 'question_digest', 'nonce', 'ceremony_digest') + } + result['statement'] = FINAL_STATEMENT + return result + + +async def challenge_at(app, key, subject, number): + started = await call(app, 'achievement.start', {}, key=key, subject=subject) + assert started.status == 'ok', wire(started) + challenge = started.data + nonces = {challenge['nonce']} + for _ in range(1, number): + step = await advance(app, key, subject, challenge) + assert step.status == 'ok', wire(step) + challenge = step.data + assert challenge['nonce'] not in nonces + nonces.add(challenge['nonce']) + assert challenge['round'] == number + return challenge + + +@pytest.mark.asyncio +@pytest.mark.parametrize('number', range(1, 6)) +@pytest.mark.parametrize( + 'fault,reason', + [('nonce', 'nonce_mismatch'), ('question', 'context_mismatch'), ('deadline', 'round_expired')], +) +async def test_each_round_rejects_context_and_can_restart(installed, number, fault, reason): + app, _ = installed + key, subject, _ = await register(app, 'honor-context') + challenge = await challenge_at(app, key, subject, number) + damaged = dict(challenge) + if fault == 'nonce': + damaged['nonce'] = ('A' if challenge['nonce'][0] != 'A' else 'B') + challenge['nonce'][1:] + elif fault == 'question': + damaged['question_digest'] = 'sha256:' + '0' * 64 + else: + # The deadline is inclusive; all previous fixture answers happened NOW. + app.executor.clock = lambda: NOW + timedelta(seconds=60) + if number == 5: + failed = await call( + app, 'achievement.finish', confirmation(damaged), key=key, subject=subject + ) + else: + failed = await advance(app, key, subject, damaged) + assert failed.status == 'ok' and failed.data['status'] == 'failed', wire(failed) + assert failed.data['reason'] == reason + async with app.metadata.transaction(write=False) as tx: + state = loads( + tx.one('SELECT body FROM achievement_ceremonies WHERE id=?', (challenge['challenge_id'],))[ + 0 + ] + ) + assert state['status'] == 'failed' + assert 'nonce_digest' not in state and 'expected_digest' not in state + assert tx.one('SELECT COUNT(*) FROM achievement_grants WHERE subject=?', (subject,))[0] == 0 + restarted = await call(app, 'achievement.start', {}, key=key, subject=subject) + assert restarted.status == 'ok' and restarted.data['round'] == 1, wire(restarted) + assert restarted.data['challenge_id'] != challenge['challenge_id'] + assert restarted.data['nonce'] != challenge['nonce'] + + +@pytest.mark.asyncio +async def test_foreign_subject_cannot_finish_or_consume_owner_ceremony(installed): + app, _ = installed + key, subject, _ = await register(app, 'honor-owner-boundary') + other_key, other, _ = await register(app, 'honor-foreign-boundary') + challenge = await challenge_at(app, key, subject, 5) + before = await business_snapshot(app) + denied = await call( + app, 'achievement.finish', confirmation(challenge), key=other_key, subject=other + ) + assert denied.status == 'error' and denied.error.code == 'achievement_ceremony_not_found' + assert await business_snapshot(app) == before + finished = await call( + app, 'achievement.finish', confirmation(challenge), key=key, subject=subject + ) + assert finished.status == 'ok', wire(finished) + + +@pytest.mark.asyncio +async def test_concurrent_finish_issues_one_grant_and_replays_only_committed_request(installed): + app, _ = installed + key, subject, _ = await register(app, 'honor-final-race') + challenge = await challenge_at(app, key, subject, 5) + args = confirmation(challenge) + request_ids = ('honor-finish-a', 'honor-finish-b') + results = await asyncio.gather( + *(call(app, 'achievement.finish', args, key=key, subject=subject, rid=rid) for rid in request_ids) + ) + assert sorted(result.status for result in results) == ['error', 'ok'] + winning = next(index for index, result in enumerate(results) if result.status == 'ok') + async with app.metadata.transaction(write=False) as tx: + assert tx.one('SELECT COUNT(*) FROM achievement_grants WHERE subject=?', (subject,))[0] == 1 + audit = [loads(row[0]) for row in tx.rows('SELECT body FROM audit')] + issued = [item for item in audit if item['event']['type'] == 'achievement.auto.issue'] + assert len(issued) == 1 + before = await business_snapshot(app) + repeated = await call( + app, 'achievement.finish', args, key=key, subject=subject, rid=request_ids[winning] + ) + assert repeated.status == 'ok' and repeated.replayed + assert repeated.data == results[winning].data + assert await business_snapshot(app) == before diff --git a/tests/test_bounty_commit_boundary.py b/tests/test_bounty_commit_boundary.py new file mode 100644 index 00000000..d8a660e4 --- /dev/null +++ b/tests/test_bounty_commit_boundary.py @@ -0,0 +1,60 @@ +"""A real deferred COMMIT failure cannot consume any Bounty business fact.""" + +import pytest +from read_only_evidence import business_snapshot +from test_market_72 import fixture, proof +from test_service import call + +from msg.core.codec import wire +from msg.plugins.money import _balance, _supply + + +@pytest.mark.asyncio +async def test_deferred_claim_commit_failure_preserves_all_rows_and_retry(installed): + app, root = installed + _, (key, buyer), created = await fixture(app, root) + listing = created.data['bounty']['listing_id'] + args = await proof(app, (key, buyer), listing) + async with app.metadata.transaction(write=True) as tx: + tx.execute( + """CREATE FUNCTION reject_test_bounty_commit() RETURNS trigger AS $$ + BEGIN RAISE EXCEPTION 'injected_bounty_commit_failure'; END; + $$ LANGUAGE plpgsql""", + write=True, + ) + tx.execute( + """CREATE CONSTRAINT TRIGGER reject_test_bounty_commit + AFTER INSERT ON bounty_claims DEFERRABLE INITIALLY DEFERRED + FOR EACH ROW EXECUTE FUNCTION reject_test_bounty_commit()""", + write=True, + ) + before = await business_snapshot(app) + failed = await call( + app, 'bounty.claim', args, key=key, subject=buyer, rid='deferred-bounty-claim' + ) + assert failed.status == 'error', wire(failed) + assert await business_snapshot(app) == before + async with app.metadata.transaction(write=True) as tx: + tx.execute('DROP TRIGGER reject_test_bounty_commit ON bounty_claims', write=True) + tx.execute('DROP FUNCTION reject_test_bounty_commit()', write=True) + paid = await call( + app, 'bounty.claim', args, key=key, subject=buyer, rid='deferred-bounty-claim' + ) + assert paid.status == 'ok' and not paid.replayed, wire(paid) + async with app.metadata.transaction(write=False) as tx: + assert _balance(tx, buyer) == 10 and _supply(tx) == 40 + assert tx.one('SELECT COUNT(*) FROM bounty_claims WHERE listing_id=?', (listing,))[0] == 1 + ledger_count = tx.one( + 'SELECT COUNT(*) FROM money_ledger WHERE reference=?', ('bounty_claim:' + listing,) + )[0] + notice_count = tx.one( + "SELECT COUNT(*) FROM messages WHERE recipient=? AND body::jsonb->>'source'='bounty_claim'", + (buyer,), + )[0] + assert ledger_count == notice_count == 1 + committed = await business_snapshot(app) + replay = await call( + app, 'bounty.claim', args, key=key, subject=buyer, rid='deferred-bounty-claim' + ) + assert replay.status == 'ok' and replay.replayed and replay.data == paid.data + assert await business_snapshot(app) == committed From 7aacc278cf26e8660a8fb7a4ca953cd6f32fd44e Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Wed, 30 Sep 2026 07:23:37 +0800 Subject: [PATCH 5/5] Complete wrong-answer, subject-limit and verifier rejection evidence --- tests/test_achievement_context_boundary.py | 39 +++++++++++++++++++--- tests/test_bounty_commit_boundary.py | 33 ++++++++++++++++++ 2 files changed, 68 insertions(+), 4 deletions(-) diff --git a/tests/test_achievement_context_boundary.py b/tests/test_achievement_context_boundary.py index bbb8a63c..647fda35 100644 --- a/tests/test_achievement_context_boundary.py +++ b/tests/test_achievement_context_boundary.py @@ -64,9 +64,9 @@ async def test_each_round_rejects_context_and_can_restart(installed, number, fau assert failed.data['reason'] == reason async with app.metadata.transaction(write=False) as tx: state = loads( - tx.one('SELECT body FROM achievement_ceremonies WHERE id=?', (challenge['challenge_id'],))[ - 0 - ] + tx.one( + 'SELECT body FROM achievement_ceremonies WHERE id=?', (challenge['challenge_id'],) + )[0] ) assert state['status'] == 'failed' assert 'nonce_digest' not in state and 'expected_digest' not in state @@ -77,6 +77,34 @@ async def test_each_round_rejects_context_and_can_restart(installed, number, fau assert restarted.data['nonce'] != challenge['nonce'] +@pytest.mark.asyncio +@pytest.mark.parametrize('number', range(1, 6)) +async def test_wrong_answer_in_any_round_ends_ceremony(installed, number): + app, _ = installed + key, subject, _ = await register(app, 'honor-wrong-answer') + challenge = await challenge_at(app, key, subject, number) + if number == 5: + failed = await call( + app, + 'achievement.finish', + {**confirmation(challenge), 'statement': 'n'}, + key=key, + subject=subject, + ) + else: + failed = await advance(app, key, subject, challenge, answer='n') + assert failed.status == 'ok' and failed.data['status'] == 'failed', wire(failed) + async with app.metadata.transaction(write=False) as tx: + assert tx.one('SELECT COUNT(*) FROM achievement_grants WHERE subject=?', (subject,))[0] == 0 + if number == 5: + continued = await call( + app, 'achievement.finish', confirmation(challenge), key=key, subject=subject + ) + else: + continued = await advance(app, key, subject, challenge) + assert continued.status == 'error' and continued.error.code == 'ceremony_inactive' + + @pytest.mark.asyncio async def test_foreign_subject_cannot_finish_or_consume_owner_ceremony(installed): app, _ = installed @@ -103,7 +131,10 @@ async def test_concurrent_finish_issues_one_grant_and_replays_only_committed_req args = confirmation(challenge) request_ids = ('honor-finish-a', 'honor-finish-b') results = await asyncio.gather( - *(call(app, 'achievement.finish', args, key=key, subject=subject, rid=rid) for rid in request_ids) + *( + call(app, 'achievement.finish', args, key=key, subject=subject, rid=rid) + for rid in request_ids + ) ) assert sorted(result.status for result in results) == ['error', 'ok'] winning = next(index for index, result in enumerate(results) if result.status == 'ok') diff --git a/tests/test_bounty_commit_boundary.py b/tests/test_bounty_commit_boundary.py index d8a660e4..d2eb2b87 100644 --- a/tests/test_bounty_commit_boundary.py +++ b/tests/test_bounty_commit_boundary.py @@ -58,3 +58,36 @@ async def test_deferred_claim_commit_failure_preserves_all_rows_and_retry(instal ) assert replay.status == 'ok' and replay.replayed and replay.data == paid.data assert await business_snapshot(app) == committed + + +@pytest.mark.asyncio +async def test_subject_limit_rejects_fresh_challenge_while_budget_remains(installed): + app, root = installed + _, (key, buyer), created = await fixture(app, root, max_claims=2) + listing = created.data['bounty']['listing_id'] + args = await proof(app, (key, buyer), listing) + paid = await call(app, 'bounty.claim', args, key=key, subject=buyer) + assert paid.status == 'ok', wire(paid) + assert paid.data['escrow_balance_minor'] == 10 + before = await business_snapshot(app) + denied = await call(app, 'bounty.challenge', {'listing_id': listing}, key=key, subject=buyer) + assert denied.status == 'error' and denied.error.code == 'bounty_subject_limit' + assert await business_snapshot(app) == before + + +@pytest.mark.asyncio +async def test_unsupported_challenge_verifier_cannot_consume_budget_or_nonce(installed): + app, root = installed + _, (key, buyer), created = await fixture(app, root) + listing = created.data['bounty']['listing_id'] + args = await proof(app, (key, buyer), listing) + async with app.metadata.transaction(write=True) as tx: + tx.execute( + 'UPDATE bounty_challenges SET verifier_version=2 WHERE id=?', + (args['challenge_id'],), + write=True, + ) + before = await business_snapshot(app) + denied = await call(app, 'bounty.claim', args, key=key, subject=buyer) + assert denied.status == 'error' and denied.error.code == 'bounty_verifier_unsupported' + assert await business_snapshot(app) == before