From 5b26c4376ee188f000200b00d5e1f8d47dee1185 Mon Sep 17 00:00:00 2001 From: Timur Gilmullin Date: Sat, 7 Nov 2026 12:45:52 +0300 Subject: [PATCH] Journal exact protocol epoch publication --- .../0024-owned-vpn-protocol-transitions.md | 27 + docs/development/wireguard-implementation.md | 36 +- ...024-owned-vpn-protocol-transitions.md.toml | 33 +- .../wireguard-implementation.md.toml | 44 +- ...024-owned-vpn-protocol-transitions.md.toml | 24 +- .../wireguard-implementation.md.toml | 31 +- src/flayer/_vpn_epoch_publication.py | 597 ++++++++++++++++++ src/flayer/_vpn_protocol_epochs.py | 34 +- src/flayer/_vpn_revisions.py | 22 + tests/unit/test_vpn_epoch_publication.py | 572 +++++++++++++++++ .../unit/test_vpn_epoch_publication_codec.py | 100 +++ 11 files changed, 1493 insertions(+), 27 deletions(-) create mode 100644 src/flayer/_vpn_epoch_publication.py create mode 100644 tests/unit/test_vpn_epoch_publication.py create mode 100644 tests/unit/test_vpn_epoch_publication_codec.py diff --git a/docs/adr/0024-owned-vpn-protocol-transitions.md b/docs/adr/0024-owned-vpn-protocol-transitions.md index 04762eb..189b735 100644 --- a/docs/adr/0024-owned-vpn-protocol-transitions.md +++ b/docs/adr/0024-owned-vpn-protocol-transitions.md @@ -384,6 +384,33 @@ proceeding, closing a failed final-unlink directory barrier. Read-only admission does not synchronize or repair metadata. This backend covers native terminal operations; schema-1 unexposed cancellation remains the separate no-op path. +## Private controller publication + +Under the existing project lock, prepare a bounded journal containing all four +expected public epoch files before requesting guest completion. Its predicted +completion is not an acknowledgment and does not publish a head. Validate the +complete old graph, closed peer segment, source/ingress manifests and capacity; +allow only that candidate's confined additional artifacts. Ordinary graph readers +continue to reject the pending journal. Missing publication authority is never +reconstructed from an orphan bundle or peer directory. + +After the coordinator supplies the exact separately authenticated guest response, +durably record `observed`, create the exact empty next peer namespace, publish the +new head and retire terminal ingress metadata and, last, the publication journal. +Retain complete known-terminal provider references; unknown dispatch cannot be +retired. Immutable public files use manifest-last exact-prefix recovery; changed +bytes, inode replacements, file-order gaps and directory membership ABA refuse. +Keep the full immutable snapshot pinned across later head and cleanup barriers. +Late snapshot drift preserves or restores the same observed protective journal; +it grants no new guest/cloud authority. Repeated completion uses the same epoch +and guest response. Read-only status repairs nothing. + +Mutating project-lock admission synchronizes the published project namespace before +any subsequent peer/guest write, closing a failed final journal-unlink directory +barrier. Read-only coordination does not synchronize. These are local publication +primitives: authenticated RPC, live attestation, cloud ordering and public CLI +admission still require integration, and no local record itself proves them. + ## Consequences and validation This requires extra bounded metadata and recovery paths, but keeps original diff --git a/docs/development/wireguard-implementation.md b/docs/development/wireguard-implementation.md index 714af82..c279637 100644 --- a/docs/development/wireguard-implementation.md +++ b/docs/development/wireguard-implementation.md @@ -756,8 +756,8 @@ Schema-1 intent loading is unchanged, and these checks need no private source bo Native epochs also preserve each unrelated protocol's preceding complete service receipt or absence proof; unchanged source selection alone cannot replace that proof. -This reader has no publication workflow or public protocol CLI yet. A separate -private guest schema-2 finalizer is described below. Existing ordinary entrypoints still reject epoch metadata until +This reader has no public protocol CLI yet. Separate private guest schema-2 +completion and controller publication primitives are described below. Existing ordinary entrypoints still reject epoch metadata until the coordinator, transport-generation handling, authenticated live attestation and recovery are integrated. The reader performs no guest, cloud or desktop operation and never infers actual runtime health from a locally authored proof record. @@ -904,7 +904,7 @@ planned identity on rollback with the same credentials/runtime and a new history link. Read-only native status reports recovery required. A stopped `unchanged` rollback cannot open input until restored; previously restored input stays unchanged. -Cloud ordering, invocation of common completion, generation-bearing epoch publication, fresh-build +Cloud ordering, invocation of common completion and epoch publication, fresh-build orchestration and public protocol CLI admission remain integration work. No live server/client or traffic acceptance is implied by these private primitives. @@ -921,7 +921,7 @@ actual kernel/service state and consumed history, exact installed file identitie terminal input journal/policy and unrelated services through a trusted observer dependency. Request JSON cannot choose that dependency. Cloud proof and local graph attestation still belong to the controller. There is no concrete AWG/VLESS observer, -RPC, controller publication workflow or public CLI integration in this primitive. +RPC or public CLI integration in this primitive. Local publication is separate below. An exact immutable schema-2 completion commits the measured historical result. Its common head is published before retiring verified input/generation/runtime @@ -940,6 +940,34 @@ does not perform this synchronization or repair. Deterministic fault tests cover publication/cleanup interruptions, incomplete records, reappearing sidecars and failed directory barriers; every service/network boundary remains synthetic. +## Journaled local epoch publication + +The private controller publisher requires caller-held project coordination. It +retains a bounded complete candidate with four public epoch files before guest +completion, distinguishing `prepared` expected bytes from an `observed` exact +authenticated response supplied by the coordinator. It verifies complete previous +graph ancestry, peer closure, sources, ingress intent and capacity. It does not +infer an acknowledgment or lost authority from an orphan artifact. Public graph +traversal continues to refuse pending publication. + +After the separate exact response, publication records `observed`, creates the +empty next peer namespace, publishes the epoch head and removes only its known +terminal ingress journal and, last, its own journal. Every prefix is recoverable +from the same retained evidence; ordered immutable file prefixes are appended, +never replaced. Unknown bytes, gaps, unexplained temporary files, stale authority, +unknown dispatch, nonempty peer segments, inode replacements and membership ABA +refuse. Complete immutable file/manifest descriptors remain pinned and verified +through later head and cleanup barriers. Late snapshot drift preserves or restores +the same observed journal before returning failure. A mutation-only project-lock +barrier persists final namespace absence before later segment or guest writes; +read-only coordination remains free of synchronization or repair. + +Historical metadata recovery uses retained public manifests when private source +bodies are lost; it never reconstructs credentials. Repeated native epochs retain +all prior segments and source files. There is no real guest/provider dispatch or +live attestation in this local backend. Protocol RPC, the coordinator and public +entrypoint integration remain #128 work. + ## Remaining integration Adding/removing a protocol on an existing guest, schema-1 migration, explicit diff --git a/docs/i18n/generated/ru/repository/docs/adr/0024-owned-vpn-protocol-transitions.md.toml b/docs/i18n/generated/ru/repository/docs/adr/0024-owned-vpn-protocol-transitions.md.toml index b28bb06..1c9978e 100644 --- a/docs/i18n/generated/ru/repository/docs/adr/0024-owned-vpn-protocol-transitions.md.toml +++ b/docs/i18n/generated/ru/repository/docs/adr/0024-owned-vpn-protocol-transitions.md.toml @@ -1,10 +1,10 @@ schemaVersion = 1 sourcePath = "docs/adr/0024-owned-vpn-protocol-transitions.md" -sourceHash = "sha256:c217ca08ac2ff6a115d354c2217cb691141d8f0f2255941f1d557a12e65ed865" -reviewedTranslationHash = "sha256:65772dad0efeffe8d3ba20fff8a80778b08d2142a11d9ef22f49a31f761942ac" +sourceHash = "sha256:69fc2e9696d9d7dbc54b378e64d1836a90ffd330c8abe04fcbdd67b2715bd67c" +reviewedTranslationHash = "sha256:73324d415942b3d94608cfd4f9707f3ffc468ddb903ea1b25952cb33254ca685" reviewer = "Codex" reviewerType = "ai" -reviewedAt = "2026-11-07T09:16:39.090557Z" +reviewedAt = "2026-11-07T09:44:22.875277Z" reviewRoles = ["editorial", "technical"] body = '''# ADR 0024: Управляемые переходы VPN-протоколов и неизменяемые эпохи @@ -275,6 +275,33 @@ vendor-CVE, закреплённой сборки инструмента и ко не синхронизирует и не исправляет метаданные. Backend покрывает нативные конечные операции; отмена схемы 1 до внешней отправки остаётся отдельным путём без изменения. +## Приватная публикация контроллера + +Под существующей блокировкой проекта до запроса завершения гостя сохраняется +ограниченный журнал с четырьмя ожидаемыми публичными файлами эпохи. Предсказанное +завершение не является подтверждением и не публикует голову. Проверяются весь прежний +граф, закрытый клиентский сегмент, манифесты источников/входа и ёмкость; разрешены лишь +ограниченные дополнительные артефакты этого кандидата. Обычный читатель графа по-прежнему +отклоняет незавершённый журнал. Потерянные полномочия публикации не восстанавливаются +из бесхозного пакета или каталога клиентов. + +После отдельного точного аутентифицированного ответа гостя координатор надёжно +записывает `observed`, создаёт точное пустое пространство следующего сегмента клиентов, +публикует новую голову и удаляет конечные входные метаданные, последним — журнал +публикации. Сохраняются полные известные конечные ссылки провайдера; неизвестную +отправку удалить нельзя. Неизменяемые публичные файлы восстанавливаются по точному +префиксу с манифестом последним. Изменённые байты, замены inode, разрывы порядка +файлов и ABA состава каталога отклоняются. Полный неизменяемый снимок закреплён до +поздних барьеров головы и очистки. Поздний дрейф сохраняет или возвращает тот же +защитный журнал observed, не выдавая новых облачных/гостевых полномочий. Повтор +использует ту же эпоху и ответ гостя. Read-only status ничего не исправляет. + +Изменяющий допуск блокировки проекта синхронизирует опубликованное пространство +проекта до следующей записи клиентов/гостя, закрывая ошибку каталожного барьера после +последнего unlink журнала. Read-only координация не синхронизирует. Это локальные +примитивы публикации: аутентифицированный RPC, живая аттестация, облачный порядок и +публичный CLI ещё требуют интеграции; сама локальная запись их не подтверждает. + ## Последствия и проверка Нужны дополнительные ограниченные метаданные и пути восстановления, но исходная принадлежность остаётся независима от желаемых слушателей, а непрерывность устройств сохраняется. Отказ при неоднозначном состоянии предпочтительнее молчаливого принятия правил или возвращения отозванного клиента. Блокировки проекта и сервера остаются advisory для внешних облачных администраторов; распределённая атомарность или исправление произвольного дрейфа не обещаются. diff --git a/docs/i18n/generated/ru/repository/docs/development/wireguard-implementation.md.toml b/docs/i18n/generated/ru/repository/docs/development/wireguard-implementation.md.toml index 36aadcb..918fe3e 100644 --- a/docs/i18n/generated/ru/repository/docs/development/wireguard-implementation.md.toml +++ b/docs/i18n/generated/ru/repository/docs/development/wireguard-implementation.md.toml @@ -1,10 +1,10 @@ schemaVersion = 1 sourcePath = "docs/development/wireguard-implementation.md" -sourceHash = "sha256:fe5355eeb5943f3cf94956bf0a032879bc72adb0653166a578404dd6a1573fa6" -reviewedTranslationHash = "sha256:09b2113bbdc20fbca8a4dcc70027f1548cd102f4199c4ac32e0048991e49b9f7" +sourceHash = "sha256:544048731e7d4cc36d5b4500d2e1b2342e157b92700b5193628b47f8f63a5073" +reviewedTranslationHash = "sha256:0a09ed24af51a7d0c6cd9f7667e89bbc79b0d2b0feb425420c748a3b65f36473" reviewer = "Codex" reviewerType = "ai" -reviewedAt = "2026-11-07T09:16:37.416707Z" +reviewedAt = "2026-11-07T09:44:20.768529Z" reviewRoles = ["editorial", "technical"] body = '''# Границы реализации стандартного WireGuard @@ -645,8 +645,8 @@ JSON операций и прямое чтение политики security gro отсутствия каждого другого протокола; один неизменный выбор источника не позволяет заменить такое свидетельство. -У читателя ещё нет процесса публикации и публичного CLI протоколов. Отдельный приватный -серверный finalizer схемы 2 описан ниже. Обычные точки входа отклоняют метаданные эпох до интеграции координатора, +У читателя ещё нет публичного CLI протоколов. Отдельные приватные примитивы завершения +гостя схемы 2 и публикации контроллера описаны ниже. Обычные точки входа отклоняют метаданные эпох до интеграции координатора, поколений протоколов, аутентифицированной живой аттестации и восстановления. Читатель не выполняет операций на сервере, в облаке или локальной сети и не выводит фактическую исправность runtime из локально составленной записи свидетельств. @@ -798,7 +798,7 @@ JSON. Восстановление окончательно выводит ст Остановленный откат `unchanged` не открывает вход до восстановления; уже восстановленные входные правила сохраняются. -Порядок облачных действий, вызов общего завершения, публикация эпох с поколениями, координация +Порядок облачных действий, вызов общего завершения и публикации эпох, координация первой сборки и публичный CLI протоколов остаются интеграционной работой. Эти приватные механизмы не означают проверку рабочего сервера/клиента или трафика. @@ -815,8 +815,8 @@ JSON. Восстановление окончательно выводит ст фактическое ядро/службу и израсходованную историю, точные идентичности установленных файлов, конечные входные журнал/политику и остальные службы через доверенный observer. JSON запроса не выбирает эту зависимость. Облачная аттестация и локальный граф остаются -обязанностью координатора. Конкретный AWG/VLESS observer, RPC, публикация координатора -и интеграция публичного CLI в этом примитиве отсутствуют. +обязанностью координатора. Конкретный AWG/VLESS observer, RPC и интеграция публичного +CLI в этом примитиве отсутствуют. Локальная публикация описана отдельно ниже. Точная неизменяемая запись схемы 2 фиксирует измеренный исторический результат. Общая голова публикуется перед удалением проверенных журналов input/generation/runtime, @@ -836,6 +836,34 @@ runtime-файлы не перечисляются и не удаляются. и очистки, неполные записи, повторное появление журналов и сбои каталожных барьеров; все сервисные/сетевые границы остаются синтетическими. +## Журналируемая локальная публикация эпохи + +Приватный publisher контроллера требует удерживать блокировку проекта. До завершения +гостя сохраняется ограниченный полный кандидат с четырьмя публичными файлами эпохи; +ожидаемые байты `prepared` отделены от точного аутентифицированного ответа `observed`, +который отдельно передаёт координатор. Проверяются весь прежний граф, замыкание +клиентов, источники, намерение входа и ёмкость. Подтверждение или потерянные полномочия +не выводятся из бесхозного артефакта. Публичный обход графа отклоняет незавершённую +публикацию. + +После отдельного точного ответа записывается `observed`, создаётся пустое пространство +следующего сегмента клиентов, публикуется голова эпохи и удаляются только известный +конечный журнал входа и последним собственный журнал. Каждый допустимый префикс +восстанавливается из тех же свидетельств; упорядоченные неизменяемые файлы дописываются, +не заменяются. Неизвестные байты, разрывы, необъяснённые временные файлы, устаревшие +полномочия, неизвестная отправка, непустые сегменты, замены inode и ABA состава +отклоняются. Дескрипторы полных неизменяемых файлов/манифеста закреплены до поздних +барьеров головы и очистки. Поздний дрейф сохраняет или возвращает тот же журнал +observed перед ошибкой. Барьер изменяющей блокировки проекта сохраняет отсутствие +конечного журнала до последующих записей сегмента/гостя. Read-only координация +остаётся без синхронизации и исправлений. + +Историческое восстановление метаданных использует сохранённые публичные манифесты +при потере приватных исходников, не восстанавливая ключи. Повторные нативные эпохи +сохраняют прежние сегменты и исходные файлы. Локальный backend не отправляет команды +на реальный сервер/провайдер и не выполняет живую аттестацию. RPC протоколов, +координатор и публичные точки входа остаются частью #128. + ## Оставшаяся интеграция Добавление/удаление протокола на существующем сервере, миграция схемы 1, явные diff --git a/docs/i18n/generated/zh-CN/repository/docs/adr/0024-owned-vpn-protocol-transitions.md.toml b/docs/i18n/generated/zh-CN/repository/docs/adr/0024-owned-vpn-protocol-transitions.md.toml index 51cec22..5ba97c9 100644 --- a/docs/i18n/generated/zh-CN/repository/docs/adr/0024-owned-vpn-protocol-transitions.md.toml +++ b/docs/i18n/generated/zh-CN/repository/docs/adr/0024-owned-vpn-protocol-transitions.md.toml @@ -1,10 +1,10 @@ schemaVersion = 1 sourcePath = "docs/adr/0024-owned-vpn-protocol-transitions.md" -sourceHash = "sha256:c217ca08ac2ff6a115d354c2217cb691141d8f0f2255941f1d557a12e65ed865" -reviewedTranslationHash = "sha256:40efa9cd347dfa93e89567e92070062e1ec0db335326422f4f66fa4c917ce00d" +sourceHash = "sha256:69fc2e9696d9d7dbc54b378e64d1836a90ffd330c8abe04fcbdd67b2715bd67c" +reviewedTranslationHash = "sha256:bed978dd66059b16e1d6564ea0c39e94917d70896488785eb3608f0062ae693f" reviewer = "Codex" reviewerType = "ai" -reviewedAt = "2026-11-07T09:16:39.090557Z" +reviewedAt = "2026-11-07T09:44:22.875277Z" reviewRoles = ["editorial", "technical"] body = '''# ADR 0024：受管 VPN 协议迁移与不可变纪元 @@ -202,6 +202,24 @@ WG 事务、根、unit、预留、历史屏障或网络变更。原直接安装 的窗口。只读准入不执行同步或修复。此后端覆盖原生终态操作；模式 1 未暴露取消仍是 单独的无操作路径。 +## 私有控制器发布 + +在现有项目锁内，请求来宾完成前保存包含四个预期公共纪元文件的有界日志。预测完成并非 +确认，也不发布头。验证完整旧图、已闭合对等端段、来源/入口清单及容量，仅允许该候选 +限定的额外制品。普通图读取器仍拒绝待处理日志。丢失的发布权限不能从孤立包或对等端 +目录重建。 + +协调器提供精确且单独认证的来宾响应后，持久化 `observed`，创建精确空的下一对等端 +命名空间，发布新头，再移除终态入口元数据，最后移除发布日志。保留完整已知终态提供方 +引用；未知发送不得清理。不可变公共文件按精确前缀恢复，清单最后写入；变更字节、inode +替换、文件次序缺口及目录成员 ABA 均拒绝。完整不可变快照持续固定至后续头及清理屏障。 +晚期快照漂移保留或恢复同一 observed 保护日志，不授予新的来宾/云权限。重复完成使用 +同一纪元及来宾响应。只读状态不执行修复。 + +项目锁的变更准入在后续对等端/来宾写入前同步已发布项目命名空间，封闭最终日志 unlink +后目录屏障失败的窗口。只读协调不执行同步。这些是本地发布原语；认证 RPC、实时证明、 +云顺序及公共 CLI 准入仍待集成，任何本地记录本身均不能证明这些事实。 + ## 后果与验证 此设计增加有界元数据及恢复路径，但使原始所有权独立于目标监听器，并保留设备连续性。拒绝模糊状态优于默默采用规则或重新激活已吊销对等端。项目和来宾锁对外部云管理员仍只是 advisory；不承诺分布式原子性，也不修复任意漂移。 diff --git a/docs/i18n/generated/zh-CN/repository/docs/development/wireguard-implementation.md.toml b/docs/i18n/generated/zh-CN/repository/docs/development/wireguard-implementation.md.toml index eb6da33..1653f10 100644 --- a/docs/i18n/generated/zh-CN/repository/docs/development/wireguard-implementation.md.toml +++ b/docs/i18n/generated/zh-CN/repository/docs/development/wireguard-implementation.md.toml @@ -1,10 +1,10 @@ schemaVersion = 1 sourcePath = "docs/development/wireguard-implementation.md" -sourceHash = "sha256:fe5355eeb5943f3cf94956bf0a032879bc72adb0653166a578404dd6a1573fa6" -reviewedTranslationHash = "sha256:1e76afe0cc3210e01c1713002f1c140456b4727b9057055ddc67d2ee9459cc76" +sourceHash = "sha256:544048731e7d4cc36d5b4500d2e1b2342e157b92700b5193628b47f8f63a5073" +reviewedTranslationHash = "sha256:c46886bbcdb144a11e0f62866d5112163776085268188c6948becb582264d945" reviewer = "Codex" reviewerType = "ai" -reviewedAt = "2026-11-07T09:16:37.416707Z" +reviewedAt = "2026-11-07T09:44:20.768529Z" reviewRoles = ["editorial", "technical"] body = '''# 标准 WireGuard 实现边界 @@ -470,7 +470,7 @@ material/server 角色摘要。必需的公共意图和请求文件仍须通过 意图加载不变，以上检查无需私有来源正文。原生纪元还精确保留每个其他协议此前的完整 服务回执或缺失证明；来源选择不变本身不能授权替换该证明。 -读取器尚无发布工作流或公共协议 CLI。下文介绍独立的私有来宾模式 2 完成器。普通入口仍拒绝纪元元数据，直到 +读取器尚无公共协议 CLI。下文介绍独立的私有来宾模式 2 完成与控制器发布原语。普通入口仍拒绝纪元元数据，直到 协调器、传输代际、认证实时证明和恢复完成集成。读取器不执行来宾、云或桌面操作，也不会 从本地编写的证据记录推断实际运行健康。 @@ -573,7 +573,7 @@ unit 及空 cgroup。部分内核状态、外来文件/unit、任务/PID 和陈 恢复永久退役旧 unit；回滚用相同凭据/运行时及新历史链恢复预先规划的新身份。只读原生 状态报告需要恢复。停止的 `unchanged` 回滚在恢复前不能开放入站；此前已恢复入站保持不变。 -云顺序、调用公共完成、带代际纪元发布、首次构建编排及公共协议 CLI 准入仍需集成。 +云顺序、调用公共完成与纪元发布、首次构建编排及公共协议 CLI 准入仍需集成。 这些私有机制不代表已完成生产服务器、客户端或流量验收。 ## 原生公共完成 @@ -585,7 +585,7 @@ unit 及空 cgroup。部分内核状态、外来文件/unit、任务/PID 和陈 首次发布持有公共 → WG 锁，验证已准备代际、实际内核/服务状态及已消耗历史、精确安装 文件身份、终态入站日志/策略，并通过可信观察器依赖验证其他服务。请求 JSON 不能选择 该依赖。云证明和本地图验证仍属于协调器。此原语尚无具体 AWG/VLESS 观察器、RPC、 -协调器发布工作流或公共 CLI 集成。 +公共 CLI 集成；本地发布在下文单独说明。 精确不可变模式 2 完成记录提交测得的历史结果。先发布公共头，再移除已验证入站/代际/ 原生旁车日志，最后移除预留。未知项、已变历史及重新出现的规范旁车会拒绝最终移除。 @@ -599,6 +599,25 @@ unit 及空 cgroup。部分内核状态、外来文件/unit、任务/PID 和陈 确定性故障测试覆盖发布/清理中断、不完整记录、重新出现的旁车和目录屏障失败； 所有服务/网络边界保持合成。 +## 带日志的本地纪元发布 + +私有控制器发布器要求调用方持有项目协调锁。来宾完成前保留包含四个公共纪元文件的 +有界完整候选，区分 `prepared` 预期字节与协调器单独提供的 `observed` 精确认证响应。 +验证完整前序图祖先、对等端闭合、来源、入口意图和容量，不从孤立制品推断确认或丢失的 +权限。公共图遍历仍拒绝待处理发布。 + +收到单独精确响应后，发布记录 `observed`，创建空的下一对等端命名空间，发布纪元头， +再仅移除已知终态入口日志，最后移除自己的日志。每个有效前缀都能从同一保留证据恢复； +有序不可变文件仅追加，不替换。未知字节、次序缺口、无法解释的临时文件、陈旧权限、 +未知发送、非空对等端段、inode 替换及成员 ABA 均拒绝。完整不可变文件/清单描述符 +持续固定并验证至后续头和清理屏障。晚期快照漂移在返回失败前保留或恢复同一 observed +日志。仅变更项目锁屏障在后续段或来宾写入前持久化最终命名空间缺失；只读协调不执行 +同步或修复。 + +私有来源正文丢失时，历史元数据恢复使用保留公共清单，绝不重建凭据。重复原生纪元 +保留所有既有段及来源文件。本地后端没有真实来宾/提供方发送或实时证明。协议 RPC、 +协调器及公共入口集成仍属于任务 #128。 + ## 尚待集成 现有来宾上的协议增删、模式 1 迁移及显式云/来宾入口覆盖层仍属于任务 #128。 diff --git a/src/flayer/_vpn_epoch_publication.py b/src/flayer/_vpn_epoch_publication.py new file mode 100644 index 0000000..c37e299 --- /dev/null +++ b/src/flayer/_vpn_epoch_publication.py @@ -0,0 +1,597 @@ +"""Publish one exact completed protocol epoch through a retained private journal, without dispatching cloud or guest operations.""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +from collections.abc import Callable, Iterator +from contextlib import ExitStack, contextmanager +from dataclasses import dataclass, field, replace +from pathlib import Path +from typing import Any + +from flayer import _vpn_protocol_epochs as epochs +from flayer._vpn_creation import BundleDigest, _Decode, _Json +from flayer._vpn_ingress_journal import DecodeJournal, IngressJournal +from flayer.guest import ReadPrivate +from flayer.profiles import artifacts +from flayer.profiles.artifacts import Artifact, ArtifactBundle +from flayer.profiles.vpn import VpnError +from flayer.vpn_project import SelectedVpnProject + +MAX_PUBLICATION_BYTES = 2 * 1024 * 1024 +_PENDING = "protocol-pending.json" +_HEAD = "protocol-head.json" +_FILES = ("epoch.json", "cloud.json", "input.json", "services.json") + + +@dataclass(frozen=True, slots=True) +class EpochPublication: + """Expected public evidence is prepared before completion; observed means an exact authenticated response was supplied separately.""" + + epoch: epochs.EpochCommit = field(repr=False) + phase: str = "prepared" + + def __post_init__(self) -> None: + """Keep the candidate closed and bounded; its predicted completion alone grants no publication authority.""" + + if type(self) is not EpochPublication or type(self.epoch) is not epochs.EpochCommit or type(self.phase) is not str or self.phase not in {"prepared", "observed"}: + raise VpnError("Epoch publication requires an exact candidate and recognized acknowledgment phase") + + epochs.EpochCommit.__post_init__(self.epoch) + + if len(_Json(self.Record())) > MAX_PUBLICATION_BYTES: + raise VpnError("Epoch publication exceeds its complete retained evidence bound") + + def Record(self) -> dict[str, Any]: + """Retain all four canonical public files without storing server payloads, paths or private credentials.""" + + bundle = self.epoch.Bundle() + + return {"schema_version": 1, "phase": self.phase, "name": bundle.name, + "files": {item.name: json.loads(item.content) for item in bundle.files}} + + +def DecodePublication(project: SelectedVpnProject, content: bytes) -> EpochPublication: + """Reject duplicate/noncanonical fields and cross-bind all evidence through the existing complete epoch codec.""" + + def Unique(pairs: list[tuple[str, object]]) -> dict[str, object]: + """Accept only one interpretation at every nested JSON boundary.""" + + result: dict[str, object] = {} + + for key, value in pairs: + if key in result: + raise ValueError + + result[key] = value + + return result + + candidate = None + + try: + if type(content) is not bytes or not 1 <= len(content) <= MAX_PUBLICATION_BYTES: + raise ValueError + + value = json.loads(content, object_pairs_hook=Unique) + + if (type(value) is not dict or set(value) != {"schema_version", "phase", "name", "files"} + or type(value["schema_version"]) is not int or value["schema_version"] != 1 + or type(value["files"]) is not dict or set(value["files"]) != set(_FILES)): + raise ValueError + + bundle = ArtifactBundle(project.vpn.identity, "server", value["name"], tuple(Artifact(name, _Json(value["files"][name]), False) for name in _FILES)) + candidate = EpochPublication(epochs.DecodeEpochBundle(bundle), value["phase"]) + + if _Json(candidate.Record()) != content: + raise ValueError + + except (ValueError, TypeError, KeyError, RecursionError, UnicodeError): + candidate = None + + if candidate is None: + raise VpnError("Epoch publication journal is malformed, foreign or noncanonical") + + return candidate + + +def LoadPublication(project: SelectedVpnProject) -> EpochPublication | None: + """Read only the canonical retained candidate; interrupted temporary publication requires explicit recovery.""" + + path = project.root / _PENDING + + return DecodePublication(project, ReadPrivate(path, limit=MAX_PUBLICATION_BYTES)) if path.exists() or path.is_symlink() else None + + +def _Directory(path: Path, descriptor: int) -> None: + """Require the pinned private directory to retain its exact public pathname throughout a durability barrier.""" + + current, linked = os.fstat(descriptor), path.lstat() + artifacts._RequireOwned(current, directory=True) + artifacts._RequireOwned(linked, directory=True) + + if (current.st_dev, current.st_ino) != (linked.st_dev, linked.st_ino): + raise VpnError("Epoch publication directory was replaced") + + +def _ExactFile(directory: int, name: str, expected: bytes, *, append: bool = False, sync: bool = False) -> None: + """Finish only an exact owned byte prefix, or verify a complete immutable file, through one pinned descriptor.""" + + flags = os.O_RDWR | os.O_APPEND | os.O_CREAT if append else os.O_RDONLY + descriptor = os.open(name, flags | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=directory) + + try: + before = os.fstat(descriptor) + artifacts._RequireOwned(before, directory=False) + content = os.pread(descriptor, len(expected) + 1, 0) + current = os.fstat(descriptor) + + if (artifacts._StatKey(before) != artifacts._StatKey(current) or len(content) != before.st_size + or (not expected.startswith(content) if append else content != expected)): + raise VpnError("Epoch publication file differs from its exact authorized bytes") + + linked = os.stat(name, dir_fd=directory, follow_symlinks=False) + + if artifacts._StatKey(before) != artifacts._StatKey(linked): + raise VpnError("Epoch publication file changed during observation") + + offset = len(content) + + while append and offset < len(expected): + written = os.write(descriptor, expected[offset:]) + + if written <= 0: + raise VpnError("Epoch publication could not finish its exact byte prefix") + + offset += written + + ready = os.fstat(descriptor) + + if sync or append: + os.fsync(descriptor) + + current, linked = os.fstat(descriptor), os.stat(name, dir_fd=directory, follow_symlinks=False) + artifacts._RequireOwned(current, directory=False) + artifacts._RequireOwned(linked, directory=False) + + if (artifacts._StatKey(current) != artifacts._StatKey(linked) or artifacts._StatKey(ready) != artifacts._StatKey(current) + or (before.st_dev, before.st_ino) != (current.st_dev, current.st_ino) + or os.pread(descriptor, len(expected) + 1, 0) != expected): + raise VpnError("Epoch publication file changed before acknowledgment") + + finally: + os.close(descriptor) + + +def _Optional(path: Path, *, limit: int = MAX_PUBLICATION_BYTES) -> bytes | None: + """Distinguish exact private bytes from true absence without treating a symlink as missing.""" + + return ReadPrivate(path, limit=limit) if path.exists() or path.is_symlink() else None + + +def _Publish(project: SelectedVpnProject, name: str, content: bytes, previous: bytes | None) -> None: + """Durably replace one fixed mutable record using only its exact authorized temporary prefix and previous bytes.""" + + if name not in {_PENDING, _HEAD} or not 1 <= len(content) <= MAX_PUBLICATION_BYTES: + raise VpnError("Unsupported epoch publication record") + + path, temporary = project.root / name, "." + name + ".new" + + with artifacts._Directory(project.root) as directory: + _Directory(project.root, directory) + + if _Optional(path) != previous: + raise VpnError("Epoch publication mutable authority changed before replacement") + + _ExactFile(directory, temporary, content, append=True) + _Directory(project.root, directory) + + if _Optional(path) != previous: + raise VpnError("Epoch publication mutable authority changed during preparation") + + os.replace(temporary, name, src_dir_fd=directory, dst_dir_fd=directory) + os.fsync(directory) + _Directory(project.root, directory) + _ExactFile(directory, name, content, sync=True) + + +def _Retire(project: SelectedVpnProject, name: str, content: bytes) -> None: + """Unlink only a fixed exact acknowledged journal, retaining any replacement or unexpected temporary file.""" + + if name not in {_PENDING, "protocol-ingress-pending.json"}: + raise VpnError("Unsupported epoch journal retirement") + + with artifacts._Directory(project.root) as directory: + _Directory(project.root, directory) + _ExactFile(directory, name, content, sync=True) + raw, receipt = artifacts._ReadSnapshot(directory, name, MAX_PUBLICATION_BYTES) + + if raw != content or (project.root / ("." + name + ".new")).exists() or (project.root / ("." + name + ".new")).is_symlink(): + raise VpnError("Epoch publication journal changed before retirement") + + artifacts._AssertFile(directory, name, receipt) + os.unlink(name, dir_fd=directory) + os.fsync(directory) + _Directory(project.root, directory) + + if (project.root / name).exists() or (project.root / name).is_symlink(): + raise VpnError("Epoch publication journal reappeared during retirement") + + +def _Head(project: SelectedVpnProject) -> dict[str, str] | None: + """Read the complete canonical local authority without inferring it from an orphan epoch bundle.""" + + raw = _Optional(project.root / _HEAD, limit=65536) + + if raw is None: + return None + + value = _Decode(raw) + pointer = epochs._Pointer(value, re.escape(project.vpn.identity.stack) + r"-ep-[0-9a-f]{16}") + + if pointer is None or _Json(pointer) != raw: + raise VpnError("Epoch publication requires an exact nonempty canonical head") + + return pointer + + +def _Ingress(project: SelectedVpnProject, epoch: epochs.EpochCommit) -> IngressJournal | None: + """Require any retained dispatch journal to match a complete known terminal original or compensating operation.""" + + path = project.root / ".protocol-ingress-pending.json.new" + + if path.exists() or path.is_symlink(): + raise VpnError("Ingress publication must be reconciled before epoch completion") + + raw = _Optional(project.root / "protocol-ingress-pending.json") + + if raw is None: + return None + + journal = DecodeJournal(raw) + metadata = epochs._Metadata(epoch.metadata, epoch.identity) + original = journal.plan == epoch.cloud.plan + operation = epoch.cloud.operation if original else epoch.cloud.restorationOperation + reference = None if operation is None else operation.reference + + if (journal.creationSha256 != metadata["creation_sha256"] or journal.previousStateSha256 != metadata["previous_state_sha256"] + or not original and journal.plan != epoch.cloud.restoration + or journal.phase != ("prepared" if reference is None else "recorded") or journal.operation != reference + or not any(item["name"] == journal.Bundle().name and item["sha256"] == BundleDigest(journal.Bundle()) for item in metadata["ingress_intents"])): + raise VpnError("Epoch publication cannot retire unknown or differently bound ingress dispatch") + + return journal + + +def _Before(project: SelectedVpnProject, publication: EpochPublication) -> epochs.ProtocolGraph: + """Validate the complete old graph plus only this candidate's confined additional artifacts and empty next segment.""" + + epoch = publication.epoch + metadata = epochs._Metadata(epoch.metadata, epoch.identity) + pointer = {"name": epoch.Bundle().name, "sha256": epoch.Digest()} + head = _Head(project) + + if (epoch.identity != project.vpn.identity or head not in (metadata["parent"], pointer) + or head == pointer and publication.phase != "observed"): + raise VpnError("Epoch publication does not continue its exact retained local head") + + graph = epochs._LoadGraph(project, metadata["parent"], staged=epoch) + epochs.RequireGraphCapacity(graph, epoch.cloud.plan) + epochs._Continue(project, graph, epoch) + epochs._IngressIntents(project, epoch) + + return graph + + +def _Bundle(project: SelectedVpnProject, epoch: epochs.EpochCommit) -> None: + """Complete manifest-last public files from their journal-bound exact prefixes; unrelated or replaced files are preserved.""" + + bundle = epoch.Bundle() + name = "server-" + bundle.name + files = [(item.name, item.content) for item in bundle.files] + [("manifest.json", artifacts._Manifest(bundle))] + + with artifacts._Directory(project.Artifacts) as parent: + _Directory(project.Artifacts, parent) + + if (project.Artifacts / ("." + name + ".lock")).exists() or (project.Artifacts / ("." + name + ".lock")).is_symlink(): + raise VpnError("Existing epoch artifact publication lock requires inspection") + + try: + os.mkdir(name, 0o700, dir_fd=parent) + + except FileExistsError: + pass + + with artifacts._Directory(project.Artifacts / name) as directory: + artifacts._AssertDirectory(parent, name, directory) + actual = set(os.listdir(directory)) + names = [filename for filename, _ in files] + + if actual != set(names[:len(actual)]): + raise VpnError("Epoch bundle contains unknown or out-of-order publication files") + + receipts = [] + + for index, (filename, content) in enumerate(files): + if index < len(actual) - 1: + _ExactFile(directory, filename, content) + + else: + _ExactFile(directory, filename, content, append=True) + + raw, receipt = artifacts._ReadSnapshot(directory, filename, artifacts.MAX_ARTIFACT_BYTES) + + if raw != content: + raise VpnError("Epoch bundle changed during exact publication") + + receipts.append((filename, receipt)) + artifacts._AssertDirectory(parent, name, directory) + stamp = artifacts._StatKey(os.fstat(directory)) + os.fsync(directory) + + if artifacts._StatKey(os.fstat(directory)) != stamp: + raise VpnError("Epoch bundle membership changed during synchronization") + + for currentName, pinned in receipts: + artifacts._AssertFile(directory, currentName, pinned) + + stamp = artifacts._StatKey(os.fstat(directory)) + os.fsync(parent) + artifacts._AssertDirectory(parent, name, directory) + + if artifacts._StatKey(os.fstat(directory)) != stamp: + raise VpnError("Epoch bundle membership changed at its parent barrier") + + for currentName, pinned in receipts: + artifacts._AssertFile(directory, currentName, pinned) + + snapshot = artifacts._ValidateManifest(directory, bundle.identity, bundle.kind, bundle.name) + + if {item.name: item.content for item in snapshot.files} != {item.name: item.content for item in bundle.files}: + raise VpnError("Published epoch bundle differs from its complete retained candidate") + + _Directory(project.Artifacts, parent) + + artifacts._SyncArtifactBundle(project.Artifacts, bundle) + + +def _Segment(project: SelectedVpnProject, epoch: epochs.EpochCommit) -> None: + """Prepare only the exact empty next peer namespace, synchronizing its directories before head publication.""" + + parent = project.root / "protocol-epochs" + + with artifacts._Directory(project.root) as root: + try: + os.mkdir("protocol-epochs", 0o700, dir_fd=root) + + except FileExistsError: + pass + + with artifacts._Directory(parent) as directory: + artifacts._AssertDirectory(root, "protocol-epochs", directory) + + try: + os.mkdir(epoch.Digest(), 0o700, dir_fd=directory) + + except FileExistsError: + pass + + with artifacts._Directory(parent / epoch.Digest()) as segment: + artifacts._AssertDirectory(directory, epoch.Digest(), segment) + + if os.listdir(segment): + raise VpnError("Unpublished next peer segment must remain empty") + + stamps = [(descriptor, artifacts._StatKey(os.fstat(descriptor))) for descriptor in (segment, directory, root)] + os.fsync(segment) + os.fsync(directory) + os.fsync(root) + artifacts._AssertDirectory(directory, epoch.Digest(), segment) + artifacts._AssertDirectory(root, "protocol-epochs", directory) + _Directory(project.root, root) + + if any(artifacts._StatKey(os.fstat(descriptor)) != stamp for descriptor, stamp in stamps) or os.listdir(segment): + raise VpnError("Epoch peer namespace membership changed during synchronization") + + +@contextmanager +def _PinnedBundle(project: SelectedVpnProject, epoch: epochs.EpochCommit) -> Iterator[Callable[[], None]]: + """Retain every verified immutable inode across later segment/head/journal barriers, not merely during bundle construction.""" + + bundle = epoch.Bundle() + name = "server-" + bundle.name + + with ExitStack() as stack: + root = stack.enter_context(artifacts._Directory(project.root)) + parent = stack.enter_context(artifacts._Directory(project.Artifacts)) + directory = stack.enter_context(artifacts._Directory(project.Artifacts / name)) + snapshot = artifacts._ValidateManifest(directory, bundle.identity, bundle.kind, bundle.name) + + if ({item.name: item.content for item in snapshot.files} != {item.name: item.content for item in bundle.files} + or snapshot.manifest.sha256 != hashlib.sha256(artifacts._Manifest(bundle)).hexdigest()): + raise VpnError("Epoch publication cannot pin a different immutable bundle") + + stamp = artifacts._StatKey(os.fstat(directory)) + pinned = [] + + for filename, receipt in [(item.name, item.receipt) for item in snapshot.files] + [("manifest.json", snapshot.manifest)]: + descriptor = os.open(filename, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) + stack.callback(os.close, descriptor) + + if artifacts._StatKey(os.fstat(descriptor)) != (receipt.device, receipt.inode, receipt.size, receipt.modifiedNs, receipt.changedNs): + raise VpnError("Epoch immutable file changed before its lifetime pin") + + pinned.append(descriptor) + + def Verify() -> None: + """Detect same-byte replacements, in-place edits and directory membership ABA before retiring publication protection.""" + + try: + _Directory(project.root, root) + _Directory(project.Artifacts, parent) + artifacts._AssertDirectory(parent, name, directory) + + if (artifacts._StatKey(os.fstat(directory)) != stamp + or artifacts._ValidateManifest(directory, bundle.identity, bundle.kind, bundle.name) != snapshot): + raise ValueError + + except (OSError, ValueError): + raise VpnError("Pinned epoch bundle changed during completed-state publication") from None + + Verify() + + for descriptor in pinned: + os.fsync(descriptor) + Verify() + + os.fsync(directory) + os.fsync(parent) + Verify() + yield Verify + Verify() + + +def _CommitPrepared( + project: SelectedVpnProject, observed: EpochPublication, current: EpochPublication, + ingress: IngressJournal | None, pointer: dict[str, str], +) -> epochs.ProtocolGraph: + """Restore only the same observed protection after snapshot drift, including final traversal and context-exit verification.""" + + root = project.root.lstat() + + try: + return _PublishPrepared(project, observed, current, ingress, pointer) + + except ValueError: + linked = project.root.lstat() + + if ((root.st_dev, root.st_ino) == (linked.st_dev, linked.st_ino) + and _Optional(project.root / _PENDING) is None and _Head(project) == pointer): + _Publish(project, _PENDING, _Json(observed.Record()), None) + + raise + + +def _PublishPrepared( + project: SelectedVpnProject, observed: EpochPublication, current: EpochPublication, + ingress: IngressJournal | None, pointer: dict[str, str], +) -> epochs.ProtocolGraph: + """Keep the complete immutable snapshot protected until final traversal and the pinned context both return successfully.""" + + with _PinnedBundle(project, observed.epoch) as verify: + _Publish(project, _PENDING, _Json(observed.Record()), _Json(current.Record())) + verify() + _Segment(project, observed.epoch) + verify() + _Before(project, observed) + graph = epochs._LoadGraph(project, pointer) + + if graph.epochs[-1] != observed.epoch or graph.revision.recordCount: + raise VpnError("Epoch publication cannot adopt a changed completed candidate or next peer segment") + + previousHead = _Head(project) + parent = epochs._Metadata(observed.epoch.metadata, observed.epoch.identity)["parent"] + + if previousHead not in (parent, pointer): + raise VpnError("Epoch publication head changed before exact replacement") + + verify() + _Publish(project, _HEAD, _Json(pointer), None if previousHead is None else _Json(previousHead)) + verify() + + if ingress is not None: + _Retire(project, "protocol-ingress-pending.json", _Json(ingress.Record())) + verify() + + _Before(project, observed) + graph = epochs._LoadGraph(project, pointer) + + if (graph.epochs[-1] != observed.epoch or graph.revision.recordCount or _Head(project) != pointer + or any(_Optional(project.root / name) is not None for name in + (".protocol-head.json.new", "protocol-ingress-pending.json", ".protocol-ingress-pending.json.new"))): + raise VpnError("Epoch publication has unexplained temporary metadata before retirement") + + verify() + + _Retire(project, _PENDING, _Json(observed.Record())) + verify() + + return epochs.LoadCommittedGraph(project) + + +def PreparePublication(project: SelectedVpnProject, epoch: epochs.EpochCommit, *, allowMutation: bool = False) -> EpochPublication: + """Under ProjectLock, retain the full expected public candidate before requesting guest completion; no dispatch or authority is granted.""" + + expected = EpochPublication(epoch) + + if allowMutation is not True: + raise VpnError("Epoch publication preparation requires explicit mutation consent") + + current = LoadPublication(project) + + if current not in (None, expected): + raise VpnError("Another or already acknowledged epoch publication requires its exact recovery") + + if (current is None and ((project.Artifacts / ("server-" + epoch.Bundle().name)).exists() + or (project.Artifacts / ("server-" + epoch.Bundle().name)).is_symlink() + or (project.root / "protocol-epochs" / epoch.Digest()).exists() + or (project.root / "protocol-epochs" / epoch.Digest()).is_symlink())): + raise VpnError("An orphan epoch or peer segment cannot reconstruct missing publication authority") + + if _Optional(project.root / ".protocol-head.json.new") is not None: + raise VpnError("Unacknowledged epoch preparation cannot explain a head publication") + + _Ingress(project, epoch) + _Before(project, expected) + raw = None if current is None else _Json(current.Record()) + _Publish(project, _PENDING, _Json(expected.Record()), raw) + _Bundle(project, epoch) + + return expected + + +def CommitPublication( + project: SelectedVpnProject, expected: EpochPublication, completion: bytes, *, allowMutation: bool = False, +) -> epochs.ProtocolGraph: + """Under ProjectLock, record the exact authenticated guest response before publishing its epoch; this function performs no live attestation.""" + + EpochPublication.__post_init__(expected) + + if allowMutation is not True or type(completion) is not bytes or completion != expected.epoch.completion: + raise VpnError("Epoch publication requires consent and the exact separately observed guest completion") + + observed = replace(expected, phase="observed") + current = LoadPublication(project) + pointer = {"name": expected.epoch.Bundle().name, "sha256": expected.epoch.Digest()} + + if current is None: + if _Head(project) != pointer: + raise VpnError("Missing epoch publication journal is never reconstructed from a candidate") + + with artifacts._Directory(project.root) as directory: + os.fsync(directory) + _Directory(project.root, directory) + + graph = epochs.LoadCommittedGraph(project) + + if graph.epochs[-1] != expected.epoch: + raise VpnError("Historical publication replay differs from the actual committed epoch") + + return graph + + if current not in (replace(expected, phase="prepared"), observed): + raise VpnError("Epoch acknowledgment differs from the retained publication candidate") + + temporaryHead = _Optional(project.root / ".protocol-head.json.new", limit=65536) + + if temporaryHead is not None and (current.phase != "observed" or not _Json(pointer).startswith(temporaryHead)): + raise VpnError("Epoch head temporary bytes differ from the exact acknowledged publication") + + ingress = _Ingress(project, expected.epoch) + _Before(project, current) + _Bundle(project, expected.epoch) + + return _CommitPrepared(project, observed, current, ingress, pointer) diff --git a/src/flayer/_vpn_protocol_epochs.py b/src/flayer/_vpn_protocol_epochs.py index 4906082..c43ba4c 100644 --- a/src/flayer/_vpn_protocol_epochs.py +++ b/src/flayer/_vpn_protocol_epochs.py @@ -999,7 +999,6 @@ def LoadCommittedGraph(project: SelectedVpnProject) -> ProtocolGraph: if path.exists() or path.is_symlink(): raise VpnError("Pending protocol metadata requires exact coordinator recovery before committed graph traversal") - baseline = LoadCreationBaseline(project) path = project.root / "protocol-head.json" raw = ReadPrivate(path, limit=65536) if path.exists() or path.is_symlink() else None head = _Pointer(_Decode(raw), re.escape(project.vpn.identity.stack) + r"-ep-[0-9a-f]{16}") if raw is not None else None @@ -1007,6 +1006,19 @@ def LoadCommittedGraph(project: SelectedVpnProject) -> ProtocolGraph: if raw is not None and (head is None or raw != _Json(head)): raise VpnError("Protocol epoch head must be its exact canonical nonempty pointer") + return _LoadGraph(project, head) + + +def _LoadGraph(project: SelectedVpnProject, head: dict[str, str] | None, *, staged: EpochCommit | None = None) -> ProtocolGraph: + """Traverse an explicit retained prefix for publication recovery; staged names are confined by one closed expected epoch.""" + + if staged is not None: + EpochCommit.__post_init__(staged) + + if staged.identity != project.vpn.identity or _Metadata(staged.metadata, staged.identity)["parent"] != head: + raise VpnError("Staged epoch cannot redirect its exact preceding graph") + + baseline = LoadCreationBaseline(project) current = head epochs = [] visited: set[str] = set() @@ -1062,16 +1074,32 @@ def LoadCommittedGraph(project: SelectedVpnProject) -> ProtocolGraph: if parent.exists() or parent.is_symlink(): PrivatePath(parent, directory=True) - if not segments or {item.name for item in parent.iterdir()} != segments: + actualSegments = {item.name for item in parent.iterdir()} + allowed = {staged.Digest()} if staged is not None else set() + + if (not segments and staged is None or segments - actualSegments or actualSegments - segments - allowed): raise VpnError("Protocol graph has unknown or missing peer segment directories") + for name in actualSegments - segments: + directory = parent / name + PrivatePath(directory, directory=True) + + if any(directory.iterdir()): + raise VpnError("Staged epoch peer segment must remain empty until its publication completes") + elif segments: raise VpnError("Protocol graph lost its retained peer segment namespace") prefixes = tuple("server-" + project.vpn.identity.stack + "-" + role + "-" for role in ("wp", "wc", "tp", "ip", "ep")) actual = {item.name.removeprefix("server-") for item in project.Artifacts.iterdir() if item.name.startswith(prefixes)} + allowedArtifacts = set() + + if staged is not None: + metadata = _Metadata(staged.metadata, staged.identity) + allowedArtifacts = {staged.Bundle().name, *(item["name"] for item in metadata["ingress_intents"]), + *(item["request_name"] for item in metadata["sources"] if item["request_name"])} - if actual - expected: + if actual - expected - allowedArtifacts: raise VpnError("Unrecorded protocol/peer/ingress artifacts require recovery; no missing head was adopted") return ProtocolGraph(baseline, graph.epochs, graph.revision, head, graph.sources, graph.activeProtocols, frozenset(expected)) diff --git a/src/flayer/_vpn_revisions.py b/src/flayer/_vpn_revisions.py index 87133b5..fde760d 100644 --- a/src/flayer/_vpn_revisions.py +++ b/src/flayer/_vpn_revisions.py @@ -152,6 +152,28 @@ def ProjectLock(project: SelectedVpnProject, *, readOnly: bool) -> Iterator[None except OSError: raise VpnError("Selected project is busy or its coordination file is unavailable") from None + if not readOnly and ((project.root / "protocol-head.json").exists() or (project.root / "protocol-head.json").is_symlink()): + directory = os.open(project.root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + + try: + opened = os.fstat(directory) + linked = project.root.lstat() + + if (not stat.S_ISDIR(opened.st_mode) or opened.st_uid != os.getuid() or stat.S_IMODE(opened.st_mode) != 0o700 + or (opened.st_dev, opened.st_ino) != (linked.st_dev, linked.st_ino)): + raise VpnError("Protocol publication namespace changed before mutating project admission") + + # Persist final journal absence before a later peer/guest write. + os.fsync(directory) + linked = project.root.lstat() + + if ((opened.st_dev, opened.st_ino) != (linked.st_dev, linked.st_ino) + or linked.st_uid != os.getuid() or stat.S_IMODE(linked.st_mode) != 0o700): + raise VpnError("Protocol publication namespace changed during mutating project admission") + + finally: + os.close(directory) + yield if descriptor is None and (path.exists() or path.is_symlink()): diff --git a/tests/unit/test_vpn_epoch_publication.py b/tests/unit/test_vpn_epoch_publication.py new file mode 100644 index 0000000..d99fa74 --- /dev/null +++ b/tests/unit/test_vpn_epoch_publication.py @@ -0,0 +1,572 @@ +"""Recover real private epoch-publication prefixes while all cloud, guest and service facts remain synthetic.""" + +from __future__ import annotations + +import json +import os +import stat +from pathlib import Path +from typing import Any + +import pytest +import test_vpn_native_graph as nativeFixtures +import test_vpn_protocol_graph as graphFixtures +from test_vpn_project_v2 import Snapshot +from test_wireguard_peer_lifecycle import Interrupted + +from flayer import _vpn_epoch_publication as publication +from flayer import _vpn_protocol_epochs as epochs +from flayer import _vpn_revisions as revisions +from flayer._vpn_creation import _Json +from flayer._vpn_ingress_journal import IngressJournal +from flayer.guest import WritePrivate +from flayer.vpn_project import SelectedVpnProject + +pytestmark = pytest.mark.skipif(os.name != "posix", reason="Private publication needs POSIX metadata/locks; no guest, cloud, WSL or desktop network is used") + + +def DispatchJournal(project: SelectedVpnProject, epoch: epochs.EpochCommit) -> IngressJournal: + """Retain a separately known terminal adapter reference without submitting or polling an actual provider operation.""" + + metadata = json.loads(epoch.metadata) + operation = epoch.cloud.operation + assert operation is not None + journal = IngressJournal(epoch.cloud.plan, metadata["creation_sha256"], metadata["previous_state_sha256"], + metadata["reservation"]["plan_expires_at"], "recorded", operation.reference) + WritePrivate(project.root / "protocol-ingress-pending.json", _Json(journal.Record())) + + return journal + + +def Environment(tmpPath: Path) -> tuple[SelectedVpnProject, epochs.EpochCommit]: + """Bind a complete synthetic native terminal result to real retained creation, source and ingress manifests.""" + + project = graphFixtures.Initial(tmpPath) + epoch = nativeFixtures.Commit(project, epochs.LoadCommittedGraph(project), "enable", "0" * 32) + DispatchJournal(project, epoch) + + return project, epoch + + +def Prepare(project: SelectedVpnProject, epoch: epochs.EpochCommit) -> publication.EpochPublication: + """Use the same project coordination required of the future authenticated controller.""" + + with revisions.ProjectLock(project, readOnly=False): + return publication.PreparePublication(project, epoch, allowMutation=True) + + +def Commit(project: SelectedVpnProject, expected: publication.EpochPublication) -> epochs.ProtocolGraph: + """Supply the fixture's separately observed exact completion; the local backend cannot observe a real guest.""" + + with revisions.ProjectLock(project, readOnly=False): + return publication.CommitPublication(project, expected, expected.epoch.completion, allowMutation=True) + + +def test_PreparedCandidateBlocksOrdinaryTraversalUntilItsSeparateAcknowledgmentPublishesOneEpoch(tmp_path: Path) -> None: + """No predicted completion is a committed graph; publication retains all source bytes and original ownership.""" + + project, epoch = Environment(tmp_path) + creation = Snapshot(project.Artifacts) + expected = Prepare(project, epoch) + assert expected.phase == "prepared" and publication.LoadPublication(project) == expected + assert not (project.root / "protocol-head.json").exists() + + with pytest.raises(ValueError, match="recovery"): + epochs.LoadCommittedGraph(project) + + repeated = Prepare(project, epoch) + assert repeated == expected + result = Commit(project, expected) + assert result.epochs == (epoch,) and result.head == {"name": epoch.Bundle().name, "sha256": epoch.Digest()} + assert result.revision.segment is not None and result.revision.segment.genesis == epoch.Digest() + assert result.revision.history == json.loads(epoch.metadata)["wireguard_history_sha256"] + assert not (project.root / "protocol-pending.json").exists() + assert not (project.root / "protocol-ingress-pending.json").exists() + after = Snapshot(project.root) + assert Commit(project, expected) == result and Snapshot(project.root) == after + assert all(Snapshot(project.Artifacts)[name] == record for name, record in creation.items()) + + +@pytest.mark.parametrize("boundary", ["pending", "epoch.json", "cloud.json", "input.json", "services.json", "manifest.json"]) +def test_PreparationRecoversEachManifestLastPublicationBoundary(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, boundary: str) -> None: + """The canonical journal precedes every public bundle file, so retry finishes only its exact ordered prefix.""" + + project, epoch = Environment(tmp_path) + publish, exact = publication._Publish, publication._ExactFile + + def Publish(*args: Any) -> None: + """Interrupt immediately after durable initial candidate publication.""" + + publish(*args) + raise Interrupted + + def Exact(directory: int, name: str, content: bytes, **kwargs: Any) -> None: + """Interrupt after one exact durable public file, leaving all private material untouched.""" + + exact(directory, name, content, **kwargs) + + if name == boundary: + raise Interrupted + + with monkeypatch.context() as fault: + fault.setattr(publication, "_Publish" if boundary == "pending" else "_ExactFile", Publish if boundary == "pending" else Exact) + + with pytest.raises(Interrupted): + Prepare(project, epoch) + + assert publication.LoadPublication(project) == publication.EpochPublication(epoch) + assert Commit(project, Prepare(project, epoch)).epochs == (epoch,) + + +@pytest.mark.parametrize("boundary", ["observed", "segment", "head", "ingress", "pending"]) +def test_RecordedAcknowledgmentRecoversEveryLocalPublicationAndJournalRetirementBoundary( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, boundary: str, +) -> None: + """A retry uses the same completion and next peer genesis; no new provider request or guest authority exists here.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + publish, segment, retire = publication._Publish, publication._Segment, publication._Retire + + def Publish(current: SelectedVpnProject, name: str, content: bytes, previous: bytes | None) -> None: + """Interrupt after acknowledgment or head publication has crossed its directory barrier.""" + + publish(current, name, content, previous) + + if boundary == "observed" and name == "protocol-pending.json" or boundary == "head" and name == "protocol-head.json": + raise Interrupted + + def Segment(*args: Any) -> None: + """Interrupt after all exact next peer-directory barriers.""" + + segment(*args) + raise Interrupted + + def Retire(current: SelectedVpnProject, name: str, content: bytes) -> None: + """Interrupt after one exact known journal is durably removed.""" + + retire(current, name, content) + + if boundary == "ingress" and name == "protocol-ingress-pending.json" or boundary == "pending" and name == "protocol-pending.json": + raise Interrupted + + with monkeypatch.context() as fault: + fault.setattr(publication, "_Publish", Publish) + fault.setattr(publication, "_Retire", Retire) + + if boundary == "segment": + fault.setattr(publication, "_Segment", Segment) + + with pytest.raises(Interrupted): + Commit(project, expected) + + result = Commit(project, expected) + assert result.epochs == (epoch,) and result.revision.segment is not None + assert result.revision.segment.genesis == epoch.Digest() + assert {path.name for path in (project.root / "protocol-epochs").iterdir()} == {epoch.Digest()} + assert not (project.root / "protocol-pending.json").exists() + + +@pytest.mark.parametrize("length", [0, 17, 300]) +def test_OnlyExactInitialJournalBytePrefixesCanContinuePreparation(tmp_path: Path, length: int) -> None: + """Partial journal bytes grant no committed graph and are finished only from the validated unchanged candidate.""" + + project, epoch = Environment(tmp_path) + record = publication.EpochPublication(epoch) + WritePrivate(project.root / ".protocol-pending.json.new", _Json(record.Record())[:length]) + + with pytest.raises(ValueError): + epochs.LoadCommittedGraph(project) + + assert Commit(project, Prepare(project, epoch)).epochs == (epoch,) + + +@pytest.mark.parametrize("filename", ["epoch.json", "cloud.json", "input.json", "services.json", "manifest.json"]) +def test_ExactPartialLastBundleFileContinuesWithoutReplacingEarlierInodes(tmp_path: Path, filename: str) -> None: + """A closed durable candidate permits append-only recovery of its last known incomplete public file.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + root = project.Artifacts / ("server-" + epoch.Bundle().name) + names = [*_FILES_FOR_TEST, "manifest.json"] + index = names.index(filename) + earlier = {name: (root / name).stat().st_ino for name in names[:index]} + + for name in names[index + 1:]: + (root / name).unlink() + + path = root / filename + path.write_bytes(path.read_bytes()[:17]) + inode = path.stat().st_ino + assert Commit(project, expected).epochs == (epoch,) + assert path.stat().st_ino == inode and all((root / name).stat().st_ino == value for name, value in earlier.items()) + + +_FILES_FOR_TEST = ("epoch.json", "cloud.json", "input.json", "services.json") + + +@pytest.mark.parametrize("fault", ["journal-lost", "wrong-response", "unknown-file", "file-gap", "bad-file", "head-temp", "ingress-unknown", "peer-drift"]) +def test_LostOrUnknownPublicationStateIsPreservedWithoutAdvancingTheHead(tmp_path: Path, fault: str) -> None: + """Complete predicted bytes cannot reconstruct lost authority, explain unknown dispatch or replace a different publication.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + bundle = project.Artifacts / ("server-" + epoch.Bundle().name) + + if fault == "journal-lost": + (project.root / "protocol-pending.json").unlink() + + elif fault == "unknown-file": + WritePrivate(bundle / "foreign.json", b"preserve unrelated bytes") + + elif fault == "file-gap": + (bundle / "cloud.json").unlink() + + elif fault == "bad-file": + (bundle / "epoch.json").write_bytes(b"preserve changed bytes") + + elif fault == "head-temp": + WritePrivate(project.root / ".protocol-head.json.new", b"preserve unknown pointer") + + elif fault == "ingress-unknown": + journal = json.loads((project.root / "protocol-ingress-pending.json").read_bytes()) + journal.update(phase="dispatching", operation=None) + (project.root / "protocol-ingress-pending.json").write_bytes(_Json(journal)) + + elif fault == "peer-drift": + directory = project.root / "protocol-epochs" + directory.mkdir(mode=0o700) + child = directory / epoch.Digest() + child.mkdir(mode=0o700) + WritePrivate(child / "wireguard-head.json", b"preserve unexplained peer authority") + + before = Snapshot(project.root) + + with revisions.ProjectLock(project, readOnly=False): + with pytest.raises(ValueError): + publication.CommitPublication(project, expected, b"wrong" if fault == "wrong-response" else epoch.completion, allowMutation=True) + + assert Snapshot(project.root) == before and not (project.root / "protocol-head.json").exists() + + if fault == "journal-lost": + with pytest.raises(ValueError, match="orphan"): + Prepare(project, epoch) + + assert Snapshot(project.root) == before + + +def test_RepeatedNativeEpochPublicationRetainsClosedSegmentsAndOriginalPrivateFiles(tmp_path: Path) -> None: + """Several enable/disable generations can complete without adopting past ingress intents or changing original credentials.""" + + project = graphFixtures.Initial(tmp_path) + original = Snapshot(project.Artifacts) + graph = epochs.LoadCommittedGraph(project) + + for index, change in enumerate(("enable", "disable", "enable", "disable")): + epoch = nativeFixtures.Commit(project, graph, change, format(index, "032x")) + DispatchJournal(project, epoch) + graph = Commit(project, Prepare(project, epoch)) + assert len(graph.epochs) == index + 1 + assert graph.revision.history == json.loads(epoch.metadata)["wireguard_history_sha256"] + assert len(list((project.root / "protocol-epochs").iterdir())) == index + 1 + + after = Snapshot(project.Artifacts) + assert all(after[name] == record for name, record in original.items()) + + +def test_PublicMetadataPublicationDoesNotRequireOrReconstructLostPrivateSourceBodies(tmp_path: Path) -> None: + """The prepared candidate, retained manifests and complete receipt pins suffice for historical metadata recovery.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + + for directory in project.Artifacts.iterdir(): + if not directory.is_dir(): + continue + + manifest = json.loads((directory / "manifest.json").read_bytes()) + + for item in manifest["files"]: + if item["sensitive"]: + (directory / item["name"]).unlink() + + before = Snapshot(project.Artifacts) + assert Commit(project, expected).epochs == (epoch,) + assert all(Snapshot(project.Artifacts)[name] == record for name, record in before.items()) + + +def test_FailedFinalUnlinkBarrierBlocksPeerSegmentMutationButKeepsReadOnlyTraversalAvailable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + """Persist root namespace absence before any later segment write, so reboot cannot resurrect an unfinishable old publication.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + root = project.root.stat() + fsync = os.fsync + events = [] + + def Fail(descriptor: int) -> None: + """Fail the final project-root removal barrier and the same barrier at subsequent mutating admission.""" + + info = os.fstat(descriptor) + + if (stat.S_ISDIR(info.st_mode) and (info.st_dev, info.st_ino) == (root.st_dev, root.st_ino) + and not (project.root / "protocol-pending.json").exists()): + raise OSError("Synthetic project publication directory barrier failure") + + fsync(descriptor) + + with monkeypatch.context() as fault: + fault.setattr(os, "fsync", Fail) + + with pytest.raises(OSError): + Commit(project, expected) + + assert not (project.root / "protocol-pending.json").exists() + + with revisions.ProjectLock(project, readOnly=True): + graph = epochs.LoadCommittedGraph(project) + assert graph.epochs == (epoch,) + + before = Snapshot(project.root) + + with pytest.raises(OSError): + with revisions.ProjectLock(project, readOnly=False): + events.append("peer mutation must remain unreachable") + + assert not events and Snapshot(project.root) == before + + def Sync(descriptor: int) -> None: + """Observe the completed root barrier before the real synthetic segment metadata write.""" + + info = os.fstat(descriptor) + fsync(descriptor) + + if (info.st_dev, info.st_ino) == (root.st_dev, root.st_ino): + events.append("durable root") + + monkeypatch.setattr(os, "fsync", Sync) + + with revisions.ProjectLock(project, readOnly=False): + assert events == ["durable root"] + revisions.AtomicRecord(project, "wireguard-pending.json", {"schema_version": 1, "synthetic": True}, segment=graph.revision.segment) + events.append("peer write") + + assert events == ["durable root", "peer write"] + + +@pytest.mark.parametrize("fault", ["file", "directory", "bundle-aba", "segment-aba"]) +def test_BundleAndPeerNamespaceReplacementOrMembershipAbaCannotPublishAHead( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fault: str, +) -> None: + """Equal bytes do not substitute for a previously verified inode; directory add/remove races also retain the protective journal.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) if fault == "segment-aba" else None + root = project.Artifacts / ("server-" + epoch.Bundle().name) + fsync = os.fsync + fired = [] + calls = [] + + def Sync(descriptor: int) -> None: + """Replace earlier public files or mutate empty directory membership at a later durability barrier.""" + + info = os.fstat(descriptor) + target = project.root / "protocol-epochs" / epoch.Digest() if fault == "segment-aba" else root + + if stat.S_ISDIR(info.st_mode) and target.exists() and (info.st_dev, info.st_ino) == (target.stat().st_dev, target.stat().st_ino): + calls.append(True) + + if not fired and (fault == "segment-aba" or len(calls) == 2): + fired.append(True) + + if fault == "file": + replacement = project.root / "synthetic-replacement.json" + WritePrivate(replacement, (root / "epoch.json").read_bytes()) + os.replace(replacement, root / "epoch.json") + + elif fault == "directory": + retained = project.Artifacts / "synthetic-retained-epoch" + root.rename(retained) + root.mkdir(mode=0o700) + + for path in retained.iterdir(): + WritePrivate(root / path.name, path.read_bytes()) + + else: + WritePrivate(target / "synthetic-membership.json", b"preserve no foreign payload") + (target / "synthetic-membership.json").unlink() + + fsync(descriptor) + + with monkeypatch.context() as injected: + injected.setattr(os, "fsync", Sync) + + with pytest.raises(ValueError): + if expected is None: + Prepare(project, epoch) + + else: + Commit(project, expected) + + assert fired and (project.root / "protocol-pending.json").exists() + assert not (project.root / "protocol-head.json").exists() + + +@pytest.mark.parametrize("entry", ["pending", "head"]) +@pytest.mark.parametrize("valid", [False, True]) +def test_UnexpectedLateMutableAuthorityCannotBecomeThePublishersExpectedPreviousBytes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, entry: str, valid: bool, +) -> None: + """Preserve a replaced pointer/journal even when its replacement has a structurally valid closed format.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + before = publication._Before + changed = [] + path = project.root / ("protocol-pending.json" if entry == "pending" else "protocol-head.json") + raw = (_Json({"name": project.vpn.identity.stack + "-ep-" + "f" * 16, "sha256": "f" * 64}) if entry == "head" + else _Json(publication.EpochPublication(epoch, "observed").Record())) if valid else b"preserve unexplained authority" + + def Before(current: SelectedVpnProject, value: publication.EpochPublication) -> epochs.ProtocolGraph: + """Replace validated authority after traversal but before the caller attempts its mutable publication.""" + + graph = before(current, value) + + if not changed and (entry == "pending" or value.phase == "observed"): + changed.append(True) + + if path.exists(): + path.write_bytes(raw) + + else: + WritePrivate(path, raw) + + return graph + + with monkeypatch.context() as fault: + fault.setattr(publication, "_Before", Before) + + with pytest.raises(ValueError): + if entry == "pending": + Prepare(project, epoch) + + else: + Commit(project, expected) + + assert changed and path.read_bytes() == raw and (project.root / "protocol-pending.json").exists() + + +def test_RecreatedTerminalIngressJournalPreventsFinalPublicationJournalRetirement(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A familiar canonical name does not explain reappearance after its verified deletion.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + retire = publication._Retire + path = project.root / "protocol-ingress-pending.json" + + def Retire(current: SelectedVpnProject, name: str, content: bytes) -> None: + """Recreate the exact acknowledged ingress journal after the cleanup wrapper returns.""" + + retire(current, name, content) + + if name == "protocol-ingress-pending.json": + WritePrivate(path, content) + + with monkeypatch.context() as fault: + fault.setattr(publication, "_Retire", Retire) + + with pytest.raises(ValueError, match="before retirement"): + Commit(project, expected) + + assert path.exists() and (project.root / "protocol-pending.json").exists() + assert Commit(project, expected).epochs == (epoch,) + + +@pytest.mark.parametrize("boundary", ["segment", "head", "ingress", "pending"]) +@pytest.mark.parametrize("replacement", ["payload", "bundle-directory"]) +def test_ImmutableBundlePinsSpanEveryLaterPublicationAndCleanupBarrier( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, boundary: str, replacement: str, +) -> None: + """A same-byte replacement after bundle synchronization cannot escape detection or leave final publication protection removed.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + bundle = project.Artifacts / ("server-" + epoch.Bundle().name) + fsync = os.fsync + fired = [] + + def Sync(descriptor: int) -> None: + """Replace immutable payload identity during a selected later directory barrier.""" + + info = os.fstat(descriptor) + target = project.root / "protocol-epochs" / epoch.Digest() if boundary == "segment" else project.root + pending = (project.root / "protocol-pending.json").exists() + ingress = (project.root / "protocol-ingress-pending.json").exists() + head = (project.root / "protocol-head.json").exists() + selected = boundary == "segment" or head and ( + boundary == "head" and pending and ingress or boundary == "ingress" and pending and not ingress or boundary == "pending" and not pending) + + if (not fired and selected and stat.S_ISDIR(info.st_mode) and target.exists() + and (info.st_dev, info.st_ino) == (target.stat().st_dev, target.stat().st_ino)): + fired.append(True) + + if replacement == "payload": + path = project.root / "synthetic-late-replacement.json" + WritePrivate(path, (bundle / "epoch.json").read_bytes()) + os.replace(path, bundle / "epoch.json") + + else: + retained = project.Artifacts / "synthetic-late-retained-epoch" + bundle.rename(retained) + bundle.mkdir(mode=0o700) + + for path in retained.iterdir(): + WritePrivate(bundle / path.name, path.read_bytes()) + + fsync(descriptor) + + with monkeypatch.context() as fault: + fault.setattr(os, "fsync", Sync) + + with pytest.raises(ValueError): + Commit(project, expected) + + assert fired and (project.root / "protocol-pending.json").exists() + assert publication.LoadPublication(project) is not None + + with pytest.raises(ValueError, match="recovery"): + epochs.LoadCommittedGraph(project) + + +def test_FinalGraphTraversalDriftStillRestoresTheSameObservedProtection(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """The restoration guard spans return-value construction and the pinned context's final verification.""" + + project, epoch = Environment(tmp_path) + expected = Prepare(project, epoch) + load = epochs.LoadCommittedGraph + replaced = [] + + def Load(current: SelectedVpnProject) -> epochs.ProtocolGraph: + """Replace an immutable inode only after final journal unlink while returning otherwise identical graph facts.""" + + result = load(current) + path = current.Artifacts / ("server-" + epoch.Bundle().name) / "epoch.json" + replacement = current.root / "synthetic-return-replacement.json" + WritePrivate(replacement, path.read_bytes()) + os.replace(replacement, path) + replaced.append(True) + + return result + + with monkeypatch.context() as fault: + fault.setattr(epochs, "LoadCommittedGraph", Load) + + with pytest.raises(ValueError, match="Pinned epoch bundle"): + Commit(project, expected) + + assert replaced and publication.LoadPublication(project) == publication.EpochPublication(epoch, "observed") + + with pytest.raises(ValueError, match="recovery"): + load(project) diff --git a/tests/unit/test_vpn_epoch_publication_codec.py b/tests/unit/test_vpn_epoch_publication_codec.py new file mode 100644 index 0000000..917d245 --- /dev/null +++ b/tests/unit/test_vpn_epoch_publication_codec.py @@ -0,0 +1,100 @@ +"""Validate retained public epoch candidates without private files, provider operations or guest execution.""" + +from __future__ import annotations + +import copy +import json +from dataclasses import replace +from types import SimpleNamespace +from typing import Any + +import pytest +from test_vpn_protocol_epochs import Commit + +from flayer import _vpn_epoch_publication as publication +from flayer._vpn_creation import _Json + + +def Candidate() -> tuple[Any, publication.EpochPublication]: + """Supply only a public synthetic complete epoch and the identity required by its closed decoder.""" + + epoch = Commit("enable") + + return SimpleNamespace(vpn=SimpleNamespace(identity=epoch.identity)), publication.EpochPublication(epoch) + + +@pytest.mark.parametrize("phase", ["prepared", "observed"]) +def test_CompletePublicCandidateRoundTripsWithoutAnImplicitObservedAcknowledgment(phase: str) -> None: + """Predicted completion remains explicitly distinguished from a separately observed response.""" + + project, value = Candidate() + value = replace(value, phase=phase) + encoded = _Json(value.Record()) + assert publication.DecodePublication(project, encoded) == value + assert json.loads(encoded)["phase"] == phase + assert set(value.Record()["files"]) == {"epoch.json", "cloud.json", "input.json", "services.json"} + + +@pytest.mark.parametrize("fault", ["schema", "extra", "phase", "name", "missing-file", "extra-file", "completion", "cloud", "source", "duplicate", "noncanonical", "oversize"]) +def test_MalformedReboundOrAmbiguousPublicationCannotReachAnyFilesystemBoundary(fault: str) -> None: + """The existing complete epoch codec validates source, intent, known cloud result and exact completion binding again.""" + + project, value = Candidate() + record = copy.deepcopy(value.Record()) + + if fault == "schema": + record["schema_version"] = True + + elif fault == "extra": + record["private_key"] = "synthetic-forbidden-field" + + elif fault == "phase": + record["phase"] = "complete" + + elif fault == "name": + record["name"] = "other-ep-" + "0" * 16 + + elif fault == "missing-file": + del record["files"]["services.json"] + + elif fault == "extra-file": + record["files"]["server.conf"] = {"forbidden": "synthetic-private-payload"} + + elif fault == "completion": + record["files"]["epoch.json"]["completion"]["intent_sha256"] = "0" * 64 + + elif fault == "cloud": + record["files"]["cloud.json"]["outcome"] = "unknown" + + elif fault == "source": + record["files"]["services.json"]["services"][0]["source_sha256"] = "0" * 64 + + encoded = _Json(record) + + if fault == "duplicate": + encoded = encoded.replace(b'"schema_version":1', b'"schema_version":1,"schema_version":1', 1) + + elif fault == "noncanonical": + encoded = json.dumps(record, indent=2).encode() + + elif fault == "oversize": + encoded = b" " * (publication.MAX_PUBLICATION_BYTES + 1) + + with pytest.raises(ValueError, match="journal"): + publication.DecodePublication(project, encoded) + + +def test_MutationConsentAndExactObservedResponseAreRequiredBeforeProjectAccess() -> None: + """Neither a predicted response nor a local observed phase authorizes guest or cloud work, or bypasses consent.""" + + _, value = Candidate() + + with pytest.raises(ValueError, match="consent"): + publication.PreparePublication(object(), value.epoch) + + for response in (value.epoch.completion, b"other-guest-response"): + with pytest.raises(ValueError, match="consent"): + publication.CommitPublication(object(), value, response) + + with pytest.raises(ValueError, match="separately observed"): + publication.CommitPublication(object(), replace(value, phase="observed"), b"other-response", allowMutation=True)