diff --git a/docs/adr/0024-owned-vpn-protocol-transitions.md b/docs/adr/0024-owned-vpn-protocol-transitions.md index c7bca9c..04762eb 100644 --- a/docs/adr/0024-owned-vpn-protocol-transitions.md +++ b/docs/adr/0024-owned-vpn-protocol-transitions.md @@ -287,8 +287,8 @@ With systemctl show --all, no job renders an empty Job value rather than a numer Cache preparation itself does not mutate an active root, unit, input policy or cloud resource. Native transitions remain private and keep the common reservation exposed. -Cloud ordering, common completion, generation-bearing epoch publication and public -CLI admission still require controller integration. Neither cache nor input preparation +Cloud ordering, invocation of common completion, generation-bearing epoch publication +and public CLI admission still require controller integration. Neither cache nor input preparation alone authorizes a service command or retires an exposed reservation. A change to another protocol preserves active WireGuard keys/configuration/root @@ -357,9 +357,33 @@ validity window and records actual restored provider IDs/tags. Reserve enough bounded recovery/epoch capacity before dispatch. A snapshot equal to the old rules does not turn an unknown submission into an undispatched cancellation. Local graph validation checks declared evidence and ancestry; authenticated live attestation, -guest completion, publication recovery and public entrypoint admission remain the +invoking guest completion, publication recovery and public entrypoint admission remain the coordinator's separate obligations. +### Private native completion + +The private common finalizer accepts a closed candidate binding the schema-2 epoch +intent, full composite request, terminal native/input journals and source-bound +service proofs. It reconstructs all native terminal records before I/O. Under the +common then WG locks, first publication rechecks pinned generation files, actual +native state/history, installed file identities and terminal input policy. A trusted +observer dependency must measure every unrelated service; caller-supplied hashes +alone cannot satisfy this requirement. Cloud and local graph attestation remain +controller responsibilities, and no RPC or public CLI invokes this backend yet. + +The exact immutable schema-2 completion is the commit point. Publish its common +head, retire only verified canonical input/generation/runtime sidecars and remove +the reservation last. Reappearing sidecars or unexplained temporary entries preserve +the reservation. A complete exact record can recover metadata publication/cleanup +after reboot without claiming current service health. An incomplete exact record +prefix requires fresh terminal attestation before append; different bytes cannot +be adopted or replaced. No recovery restores consumed peer history. + +Ordinary mutating WG admission synchronizes the completed common namespace before +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. + ## 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 e95e23e..714af82 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, guest schema-2 finalizer or public -protocol CLI yet. Existing ordinary entrypoints still reject epoch metadata until +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 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,10 +904,42 @@ 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, common completion, generation-bearing epoch publication, fresh-build +Cloud ordering, invocation of common completion, generation-bearing 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. +## Native common completion + +The private finalizer validates its closed candidate before guest I/O or mutation: +schema-2 epoch metadata, exact full composite, terminal native journal, applied or +rolled-back input proof and source-bound service receipts/absence. It supports +enabled, removed, restored, unchanged and rolled-back native outcomes. This is not +an unstarted-operation producer; unexposed schema-1 cancellation remains separate. + +First publication holds common then WG locks and verifies the prepared generation, +actual kernel/service state and consumed history, exact installed file identities, +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. + +An exact immutable schema-2 completion commits the measured historical result. +Its common head is published before retiring verified input/generation/runtime +sidecars and, last, the reservation. Unknown entries, changed history and reappearing +canonical sidecars refuse that final retirement. Cleanup pins file/directory +identities, verifies bytes and synchronizes every barrier; it does not enumerate or +delete private runtime payloads. A complete exact record permits metadata recovery +without live service health after a crash or reboot. This replay makes no current +health or peer-history claim. Only an exact incomplete byte prefix can be appended, +and only after all live terminal proofs are rechecked. Neither path rewinds history. + +If final reservation unlink succeeds but its directory synchronization fails, the +next ordinary mutating WG guest/peer entrypoint must synchronize the completed +namespace before any write. A failed barrier refuses admission. Read-only status +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. + ## 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 c9f9c2e..b28bb06 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:81e2df46a70f3781a4060976d2fd386fbceeb6b0d25651d8c368dc5cabdc65b6" -reviewedTranslationHash = "sha256:375ff603cbefdc60e8b53a8b1ed51244d22eecbd61ad2d0159de1fa91399ddff" +sourceHash = "sha256:c217ca08ac2ff6a115d354c2217cb691141d8f0f2255941f1d557a12e65ed865" +reviewedTranslationHash = "sha256:65772dad0efeffe8d3ba20fff8a80778b08d2142a11d9ef22f49a31f761942ac" reviewer = "Codex" reviewerType = "ai" -reviewedAt = "2026-11-07T08:00:48.180642Z" +reviewedAt = "2026-11-07T09:16:39.090557Z" reviewRoles = ["editorial", "technical"] body = '''# ADR 0024: Управляемые переходы VPN-протоколов и неизменяемые эпохи @@ -158,7 +158,7 @@ helper или кандидат. Неизвестный ABI отклоняетс Сама подготовка кэша не меняет активный корень, unit, входящие правила или облако. Native-переходы приватны и оставляют общую резервацию открытой. Порядок облачных действий, -общее завершение, публикация эпох с поколениями и публичный CLI ещё требуют интеграции +вызов общего завершения, публикация эпох с поколениями и публичный CLI ещё требуют интеграции контроллера. Одна подготовка кэша или входа не разрешает команду сервиса и не закрывает открытую резервацию. @@ -248,9 +248,33 @@ vendor-CVE, закреплённой сборки инструмента и ко и метки провайдера. До отправки резервируется ёмкость полного восстановления и эпохи. Совпадение снимка с прежними правилами не превращает неизвестную отправку в отмену до отправки. Локальный граф проверяет объявленные свидетельства и происхождение; -аутентифицированная живая аттестация, завершение гостя, восстановление публикации и +аутентифицированная живая аттестация, вызов завершения гостя, восстановление публикации и допуск публичных точек входа остаются отдельными обязанностями координатора. +### Приватное завершение нативной операции + +Приватный общий finalizer принимает закрытый кандидат, связывающий намерение эпохи +схемы 2, полный составной запрос, конечные журналы native/input и сервисные +свидетельства с источниками. До ввода-вывода восстанавливаются все нативные конечные +записи. Под блокировками общая → WG первая публикация повторно проверяет закреплённые +файлы поколения, фактическое нативное состояние и историю, идентичности установленных +файлов и конечную входную политику. Доверенный observer обязан измерить остальные +службы; одних заданных вызывающим кодом хешей недостаточно. Облако и локальный граф +аттестует координатор. RPC и публичный CLI пока не вызывают этот backend. + +Точная неизменяемая запись завершения схемы 2 — точка фиксации. Затем публикуется общая +голова, удаляются только проверенные канонические журналы input/generation/runtime, +последней — резервация. Повторно появившиеся журналы или необъяснённые временные +файлы сохраняют резервацию. Полная точная запись позволяет после перезагрузки +восстановить публикацию и очистку метаданных без утверждения о текущей исправности +службы. Неполный точный префикс требует свежей конечной аттестации перед дописыванием; +другие байты не принимаются и не заменяются. Израсходованная история не возвращается. + +Обычный изменяющий вход WG синхронизирует завершённое общее пространство имён перед +работой, закрывая сбой каталожного барьера после последнего unlink. Read-only допуск +не синхронизирует и не исправляет метаданные. Backend покрывает нативные конечные +операции; отмена схемы 1 до внешней отправки остаётся отдельным путём без изменения. + ## Последствия и проверка Нужны дополнительные ограниченные метаданные и пути восстановления, но исходная принадлежность остаётся независима от желаемых слушателей, а непрерывность устройств сохраняется. Отказ при неоднозначном состоянии предпочтительнее молчаливого принятия правил или возвращения отозванного клиента. Блокировки проекта и сервера остаются 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 2863ef2..36aadcb 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:638bbd1a88db8547510debfc0af295a9fddd9158324fb3f1c0f2ec38f5f27ed6" -reviewedTranslationHash = "sha256:ece20b98bec793849d21b399b8a645d7c1bdd97ababa60cd3fbab65b7d3cd47d" +sourceHash = "sha256:fe5355eeb5943f3cf94956bf0a032879bc72adb0653166a578404dd6a1573fa6" +reviewedTranslationHash = "sha256:09b2113bbdc20fbca8a4dcc70027f1548cd102f4199c4ac32e0048991e49b9f7" reviewer = "Codex" reviewerType = "ai" -reviewedAt = "2026-11-07T08:00:47.331236Z" +reviewedAt = "2026-11-07T09:16:37.416707Z" reviewRoles = ["editorial", "technical"] body = '''# Границы реализации стандартного WireGuard @@ -645,8 +645,8 @@ JSON операций и прямое чтение политики security gro отсутствия каждого другого протокола; один неизменный выбор источника не позволяет заменить такое свидетельство. -У читателя ещё нет процесса публикации, серверного finalizer схемы 2 и публичного CLI -протоколов. Обычные точки входа отклоняют метаданные эпох до интеграции координатора, +У читателя ещё нет процесса публикации и публичного CLI протоколов. Отдельный приватный +серверный finalizer схемы 2 описан ниже. Обычные точки входа отклоняют метаданные эпох до интеграции координатора, поколений протоколов, аутентифицированной живой аттестации и восстановления. Читатель не выполняет операций на сервере, в облаке или локальной сети и не выводит фактическую исправность runtime из локально составленной записи свидетельств. @@ -798,10 +798,44 @@ JSON. Восстановление окончательно выводит ст Остановленный откат `unchanged` не открывает вход до восстановления; уже восстановленные входные правила сохраняются. -Порядок облачных действий, общее завершение, публикация эпох с поколениями, координация +Порядок облачных действий, вызов общего завершения, публикация эпох с поколениями, координация первой сборки и публичный CLI протоколов остаются интеграционной работой. Эти приватные механизмы не означают проверку рабочего сервера/клиента или трафика. +## Нативное общее завершение + +Приватный finalizer проверяет закрытый кандидат до серверного ввода-вывода и изменений: +метаданные эпохи схемы 2, точный полный составной запрос, конечный native-журнал, +свидетельство применённых или восстановленных входных правил и связанные с источником +сервисные квитанции/отсутствие. Поддерживаются enabled, removed, restored, unchanged +и rolled-back нативные исходы. Это не генератор неначатых операций; отмена схемы 1 +до внешней отправки остаётся отдельной. + +Первая публикация под блокировками общая → WG проверяет подготовленное поколение, +фактическое ядро/службу и израсходованную историю, точные идентичности установленных +файлов, конечные входные журнал/политику и остальные службы через доверенный observer. +JSON запроса не выбирает эту зависимость. Облачная аттестация и локальный граф остаются +обязанностью координатора. Конкретный AWG/VLESS observer, RPC, публикация координатора +и интеграция публичного CLI в этом примитиве отсутствуют. + +Точная неизменяемая запись схемы 2 фиксирует измеренный исторический результат. +Общая голова публикуется перед удалением проверенных журналов input/generation/runtime, +а резервация — последней. Неизвестные записи, изменившаяся история и повторно +появившиеся канонические журналы не дают удалить резервацию. Очистка закрепляет +идентичности файлов/каталогов, проверяет байты и синхронизирует каждый барьер; приватные +runtime-файлы не перечисляются и не удаляются. Полная точная запись позволяет +восстановить метаданные после сбоя/перезагрузки без живой проверки службы. Повтор +не заявляет текущую исправность или текущую историю клиентов. Дописать можно только +точный неполный префикс после повторной проверки всех живых конечных свидетельств. +Ни один путь не возвращает прежнюю историю. + +Если последний unlink резервации прошёл, но синхронизация каталога завершилась ошибкой, +следующий обычный изменяющий вход WG guest/peer синхронизирует завершённое пространство +имён до любой записи. Неудачный барьер отклоняет допуск. Read-only status не выполняет +эту синхронизацию или исправление. Детерминированные тесты покрывают прерывания публикации +и очистки, неполные записи, повторное появление журналов и сбои каталожных барьеров; +все сервисные/сетевые границы остаются синтетическими. + ## Оставшаяся интеграция Добавление/удаление протокола на существующем сервере, миграция схемы 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 abc66b1..51cec22 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:81e2df46a70f3781a4060976d2fd386fbceeb6b0d25651d8c368dc5cabdc65b6" -reviewedTranslationHash = "sha256:9b797a83c6fe294eaf7a0acfddaf9653e18f05b393f2c53b73e45f019288447c" +sourceHash = "sha256:c217ca08ac2ff6a115d354c2217cb691141d8f0f2255941f1d557a12e65ed865" +reviewedTranslationHash = "sha256:40efa9cd347dfa93e89567e92070062e1ec0db335326422f4f66fa4c917ce00d" reviewer = "Codex" reviewerType = "ai" -reviewedAt = "2026-11-07T08:00:48.180642Z" +reviewedAt = "2026-11-07T09:16:39.090557Z" reviewRoles = ["editorial", "technical"] body = '''# ADR 0024：受管 VPN 协议迁移与不可变纪元 @@ -126,7 +126,7 @@ unit。这既不声明健康，也不推断发生重启。部分或外来状态 systemctl show --all 将无任务表示为空 Job 值，而不是数字零。 缓存准备本身不变更活动根、unit、入站策略或云资源。原生转换仍为私有，并保持公共预留 -暴露。云操作顺序、公共完成、带代际的纪元发布及公共 CLI 准入仍需控制器集成。仅缓存或 +暴露。云操作顺序、调用公共完成、带代际的纪元发布及公共 CLI 准入仍需控制器集成。仅缓存或 入站准备均不能授权服务命令或结束已暴露预留。 其他协议变更保留活动 WireGuard 的密钥、配置和根回执，但首次协议写入前必须在同一现有 WireGuard 锁内消耗其对等端历史授权。采用现有有效的前后回执相等的取消审计格式发布新链式历史头；已排队的旧对等端驱动字节会发现预期历史不同并拒绝变更。只有新代码理解的预留机制不足以做到这一点。此证明之前必须拒绝待处理对等端事务。将新历史写入下一纪元；若协议变更随后回滚，也要发布保持旧活动固定值/访问语义、更新完整云回执并记录已消耗历史的无活动变更纪元完成记录。回滚绝不恢复旧历史。此操作要求活动 WireGuard 的精确健康回执。若 WireGuard 不活动或不存在，则验证独立的完整命名空间缺失及保留历史证明，绝不能仅为撤销授权而伪造运行中回执。 @@ -183,7 +183,24 @@ WG 事务、根、unit、预留、历史屏障或网络变更。原直接安装 保留完整已知终态云操作和读回，以及每个原始或补偿入站意图制品。补偿绑定独立有效窗口， 记录实际恢复的提供方 ID 和标签。发送前预留完整恢复及纪元容量。与旧规则相同的快照不能 把未知提交变成未发送取消。本地图验证声明的证据及祖先关系；经过认证的实时证明、来宾 -完成、发布恢复及公共入口准入仍是协调器的独立责任。 +完成的调用、发布恢复及公共入口准入仍是协调器的独立责任。 + +### 私有原生完成 + +私有公共完成器接受封闭候选，绑定模式 2 纪元意图、完整复合请求、原生/入站终态日志及 +与来源绑定的服务证明。I/O 前重建所有原生终态记录。在公共 → WG 锁内，首次发布重新 +验证固定代际文件、实际原生状态/历史、已安装文件身份及终态入站策略。可信观察器依赖 +必须测量每项其他服务；调用方给出的摘要不足以满足要求。云和本地图证明仍由协调器负责， +目前没有 RPC 或公共 CLI 调用此后端。 + +精确不可变模式 2 完成记录是提交点。先发布公共头，再仅移除已验证的规范入站/代际/原生 +旁车日志，最后移除预留。重新出现的旁车或无法解释的临时项会保留预留。完整精确记录 +允许在重启后恢复元数据发布/清理，不声称当前服务健康。精确但不完整的字节前缀必须重新 +验证实时终态证明才能追加；不同字节不能接管或替换。恢复绝不倒退已消耗的对等端历史。 + +普通 WG 变更入口继续前必须同步已完成公共命名空间，封闭最终 unlink 后目录屏障失败 +的窗口。只读准入不执行同步或修复。此后端覆盖原生终态操作；模式 1 未暴露取消仍是 +单独的无操作路径。 ## 后果与验证 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 c613a18..eb6da33 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:638bbd1a88db8547510debfc0af295a9fddd9158324fb3f1c0f2ec38f5f27ed6" -reviewedTranslationHash = "sha256:750a2a1b6e40e9d20648bb6e788e49ce78f26ba13ca75acaa6c8dbe1a6be6749" +sourceHash = "sha256:fe5355eeb5943f3cf94956bf0a032879bc72adb0653166a578404dd6a1573fa6" +reviewedTranslationHash = "sha256:1e76afe0cc3210e01c1713002f1c140456b4727b9057055ddc67d2ee9459cc76" reviewer = "Codex" reviewerType = "ai" -reviewedAt = "2026-11-07T08:00:47.331236Z" +reviewedAt = "2026-11-07T09:16:37.416707Z" reviewRoles = ["editorial", "technical"] body = '''# 标准 WireGuard 实现边界 @@ -470,7 +470,7 @@ material/server 角色摘要。必需的公共意图和请求文件仍须通过 意图加载不变，以上检查无需私有来源正文。原生纪元还精确保留每个其他协议此前的完整 服务回执或缺失证明；来源选择不变本身不能授权替换该证明。 -读取器尚无发布工作流、来宾模式 2 完成器或公共协议 CLI。普通入口仍拒绝纪元元数据，直到 +读取器尚无发布工作流或公共协议 CLI。下文介绍独立的私有来宾模式 2 完成器。普通入口仍拒绝纪元元数据，直到 协调器、传输代际、认证实时证明和恢复完成集成。读取器不执行来宾、云或桌面操作，也不会 从本地编写的证据记录推断实际运行健康。 @@ -573,9 +573,32 @@ unit 及空 cgroup。部分内核状态、外来文件/unit、任务/PID 和陈 恢复永久退役旧 unit；回滚用相同凭据/运行时及新历史链恢复预先规划的新身份。只读原生 状态报告需要恢复。停止的 `unchanged` 回滚在恢复前不能开放入站；此前已恢复入站保持不变。 -云顺序、公共完成、带代际纪元发布、首次构建编排及公共协议 CLI 准入仍需集成。 +云顺序、调用公共完成、带代际纪元发布、首次构建编排及公共协议 CLI 准入仍需集成。 这些私有机制不代表已完成生产服务器、客户端或流量验收。 +## 原生公共完成 + +私有完成器在来宾 I/O 或变更前验证封闭候选：模式 2 纪元元数据、精确完整复合请求、 +原生终态日志、已应用或已回滚入站证明及来源绑定服务回执/缺失。支持启用、移除、恢复、 +未改变及已回滚原生结果。它不生成未启动操作；未暴露模式 1 取消仍独立处理。 + +首次发布持有公共 → WG 锁，验证已准备代际、实际内核/服务状态及已消耗历史、精确安装 +文件身份、终态入站日志/策略，并通过可信观察器依赖验证其他服务。请求 JSON 不能选择 +该依赖。云证明和本地图验证仍属于协调器。此原语尚无具体 AWG/VLESS 观察器、RPC、 +协调器发布工作流或公共 CLI 集成。 + +精确不可变模式 2 完成记录提交测得的历史结果。先发布公共头，再移除已验证入站/代际/ +原生旁车日志，最后移除预留。未知项、已变历史及重新出现的规范旁车会拒绝最终移除。 +清理固定文件/目录身份、验证字节并同步每个屏障，不枚举或删除私有运行时载荷。 +完整精确记录允许崩溃或重启后恢复元数据，不要求服务实时健康。此重放不声称当前健康 +或对等端历史。只有精确不完整字节前缀可以追加，而且必须重新检查全部实时终态证明。 +任何路径都不会倒退历史。 + +若最终预留 unlink 成功但目录同步失败，下一次普通 WG 来宾/对等端变更入口必须在任何 +写入前同步完成后的命名空间。屏障失败会拒绝准入。只读状态不执行此同步或修复。 +确定性故障测试覆盖发布/清理中断、不完整记录、重新出现的旁车和目录屏障失败； +所有服务/网络边界保持合成。 + ## 尚待集成 现有来宾上的协议增删、模式 1 迁移及显式云/来宾入口覆盖层仍属于任务 #128。 diff --git a/src/flayer/profiles/_host_input_guest.py b/src/flayer/profiles/_host_input_guest.py index b6efdd9..b11e337 100644 --- a/src/flayer/profiles/_host_input_guest.py +++ b/src/flayer/profiles/_host_input_guest.py @@ -60,6 +60,10 @@ def _Reservation( raise InputError("Host input mutation requires the exact current exposed reservation") root = authority._Root(guest) + completion = root / ("done-" + intent["operation_id"] + ".json") + + if completion.exists() or completion.is_symlink(): + raise InputError("Native completion publication requires finalization before host-input mutation") if any((root / name).exists() or (root / name).is_symlink() for name in ("runtime.json", ".runtime.json.new")): if request["schema_version"] != 2 or generationCache is None or runtimeBackend is None: diff --git a/src/flayer/profiles/_protocol_completion.py b/src/flayer/profiles/_protocol_completion.py new file mode 100644 index 0000000..23c00f4 --- /dev/null +++ b/src/flayer/profiles/_protocol_completion.py @@ -0,0 +1,440 @@ +"""Bind verified native/input/service facts to an immutable common completion before retiring its reservation.""" + +from __future__ import annotations + +import os +import re +import stat +from contextlib import ExitStack, nullcontext +from typing import Any + +_PROTOCOLS = ("amneziawg", "vless-reality", "wireguard") +_METADATA = {"schema_version", "identity", "creation_sha256", "parent", "closed_peer_head", "closed_peer_head_sha256", "closed_peer_count", + "previous_state_sha256", "sources", "configured_listeners", "active_protocols", "static_fingerprint", "wireguard_receipt_sha256", + "wireguard_history_sha256", "generation_receipt_sha256", "reservation", "authority_parent", "ingress_intents", "cloud_sha256", + "input_sha256", "services_sha256", "generation", "native_sha256"} + + +class CompletionError(ValueError): + """A bounded completion diagnostic without private runtime/configuration bytes or native output.""" + + +def _Pointer(authority: Any, value: object, pattern: str) -> dict[str, str] | None: + """Validate public graph locators without interpreting any locator as a guest filesystem path.""" + + if value is None: + return None + + if (type(value) is not dict or set(value) != {"name", "sha256"} or type(value["name"]) is not str + or re.fullmatch(pattern, value["name"]) is None or not authority._Hash(value["sha256"])): + raise ValueError + + return dict(value) + + +def _Sources(guest: Any, authority: Any, identity: dict[str, str], values: object) -> list[dict[str, Any]]: + """Retain complete source identity and confined role references, independently of any private projection bodies.""" + + if type(values) is not list or not 1 <= len(values) <= 3: + raise ValueError + + result = [] + + for source in values: + if (type(source) is not dict or set(source) != {"schema_version", "identity", "pin", "kind", "request_name", "request_sha256"} + or type(source["schema_version"]) is not int or source["schema_version"] != 1 + or guest._Encode(source["identity"]) != guest._Encode(identity) or type(source["pin"]) is not dict + or set(source["pin"]) != {"protocol", "intent_fingerprint", "material_digest", "server_digest"} + or type(source["pin"]["protocol"]) is not str or source["pin"]["protocol"] not in _PROTOCOLS + or any(not authority._Hash(source["pin"][key]) for key in ("intent_fingerprint", "material_digest", "server_digest")) + or type(source["kind"]) is not str or source["kind"] not in {"initial", "peer", "protocol"} + or type(source["request_name"]) is not str or type(source["request_sha256"]) is not str): + raise ValueError + + if source["kind"] == "initial": + if source["request_name"] or source["request_sha256"]: + raise ValueError + + else: + prefix = "wp" if source["kind"] == "peer" else "tp" + + if (source["kind"] == "peer" and source["pin"]["protocol"] != "wireguard" + or len(source["request_name"]) > 63 or not authority._Hash(source["request_sha256"]) + or re.fullmatch(re.escape(identity["stack"]) + "-" + prefix + r"-[0-9a-f]{16}", source["request_name"]) is None): + raise ValueError + + result.append(source) + + names = [source["pin"]["protocol"] for source in result] + + if names != [name for name in _PROTOCOLS if name in names]: + raise ValueError + + return result + + +def Candidate(guest: Any, authority: Any, policy: Any, cache: Any, value: object, rawIntent: object) -> dict[str, Any]: + """Cross-bind a closed native intent and terminal witness envelope before any guest I/O; provider attestation belongs to the controller.""" + + result = None + + try: + intent = authority.Intent(rawIntent) + + if (type(value) is not dict or set(value) != {"schema_version", "metadata", "request", "input", "services", "outcome"} + or type(value["schema_version"]) is not int or value["schema_version"] != 1 + or type(value["outcome"]) is not str or value["outcome"] not in {"committed", "not-applied"} + or len(guest._Encode(value)) > 131072): + raise ValueError + + request = cache.Request(guest, policy, value["request"], intent) + values = policy.Transition(request["input"]) + metadata, inputProof, serviceProof = value["metadata"], value["input"], value["services"] + + if (type(metadata) is not dict or set(metadata) != _METADATA or type(metadata["schema_version"]) is not int or metadata["schema_version"] != 2 + or len(guest._Encode(metadata)) > 32768 or request["generation"]["schema_version"] != 3 + or guest._Encode(metadata["reservation"]) != guest._Encode(intent) or guest._Encode(metadata["generation"]) != guest._Encode(request["generation"]) + or guest._Encode(metadata["identity"]) != guest._Encode(values[0]["identity"]) + or metadata["creation_sha256"] != intent["creation_sha256"] or metadata["previous_state_sha256"] != intent["previous_state_sha256"] + or any(not authority._Hash(metadata[key]) for key in ("creation_sha256", "closed_peer_head_sha256", "previous_state_sha256", "static_fingerprint", + "wireguard_history_sha256", "cloud_sha256", "input_sha256", "services_sha256", "native_sha256")) + or any(type(metadata[key]) is not str or metadata[key] != "" and not authority._Hash(metadata[key]) + for key in ("wireguard_receipt_sha256", "generation_receipt_sha256")) + or type(metadata["closed_peer_count"]) is not int or not 0 <= metadata["closed_peer_count"] <= 4096): + raise ValueError + + stack = re.escape(values[0]["identity"]["stack"]) + parent = _Pointer(authority, metadata["parent"], stack + r"-ep-[0-9a-f]{16}") + peer = _Pointer(authority, metadata["closed_peer_head"], stack + r"-wc-[0-9a-f]{16}") + authority._Pointer(metadata["authority_parent"]) + + if (bool(peer) != bool(metadata["closed_peer_count"]) + or metadata["closed_peer_head_sha256"] != (guest._Digest(guest._Encode(peer)) if peer is not None + else parent["sha256"] if parent is not None else metadata["creation_sha256"]) + or type(metadata["ingress_intents"]) is not list or len(metadata["ingress_intents"]) > 2): + raise ValueError + + roles = [] + + for reference in metadata["ingress_intents"]: + if type(reference) is not dict or set(reference) != {"role", "name", "sha256"} or reference["role"] not in {"change", "restoration"}: + raise ValueError + + _Pointer(authority, {key: reference[key] for key in ("name", "sha256")}, stack + r"-ip-[0-9a-f]{16}") + roles.append(reference["role"]) + + if len(set(roles)) != len(roles): + raise ValueError + + sources = _Sources(guest, authority, values[0]["identity"], metadata["sources"]) + names = [source["pin"]["protocol"] for source in sources] + + if (type(metadata["configured_listeners"]) is not list or [policy.Listener(item)["transport"] for item in metadata["configured_listeners"]] != names + or type(metadata["active_protocols"]) is not list or any(type(item) is not str for item in metadata["active_protocols"]) + or metadata["active_protocols"] != [name for name in names if name in metadata["active_protocols"]] + or type(serviceProof) is not dict or set(serviceProof) != {"schema_version", "services"} + or type(serviceProof["schema_version"]) is not int or serviceProof["schema_version"] != 1 + or type(serviceProof["services"]) is not list or len(serviceProof["services"]) != len(sources)): + raise ValueError + + active = metadata["active_protocols"] + committed = value["outcome"] == "committed" + targetPolicy = values[1 if committed else 2] + configured = {item["transport"]: item for item in metadata["configured_listeners"]} + + if guest._Encode({item["transport"]: item for item in targetPolicy["listeners"]}) != guest._Encode({name: configured[name] for name in active}): + raise ValueError + + for proof, source in zip(serviceProof["services"], sources): + if (type(proof) is not dict or set(proof) != {"protocol", "source_sha256", "state", "receipt_sha256", "absence_sha256"} + or proof["protocol"] != source["pin"]["protocol"] or proof["source_sha256"] != source["pin"]["server_digest"] + or proof["state"] != ("active" if proof["protocol"] in active else "absent") + or type(proof["receipt_sha256"]) is not str or type(proof["absence_sha256"]) is not str + or proof["state"] == "active" and (not authority._Hash(proof["receipt_sha256"]) or proof["absence_sha256"] != "") + or proof["state"] == "absent" and (proof["receipt_sha256"] != "" or not authority._Hash(proof["absence_sha256"]))): + raise ValueError + + if (type(inputProof) is not dict or set(inputProof) != {"schema_version", "transition", "outcome", "native_sha256", "journal_sha256", "boot_sha256"} + or type(inputProof["schema_version"]) is not int or inputProof["schema_version"] != 1 + or guest._Encode(inputProof["transition"]) != guest._Encode(request["input"]) + or inputProof["outcome"] != ("applied" if committed else "rolled-back") + or any(not authority._Hash(inputProof[key]) for key in ("native_sha256", "journal_sha256", "boot_sha256")) + or inputProof["boot_sha256"] != policy.Digest(policy.Boot(targetPolicy)) + or metadata["input_sha256"] != guest._Digest(guest._Encode(inputProof)) + or metadata["services_sha256"] != guest._Digest(guest._Encode(serviceProof))): + raise ValueError + + generation = request["generation"] + source = next(item for item in sources if item["pin"]["protocol"] == "wireguard") + fresh = guest._Digest(guest._Encode({**generation["candidate_receipt"], "status": "installed"})) + original = guest._ProtocolHistory(intent, request) + change = generation["change"] + phase = (("active" if change == "enable" else "inactive") if committed + else "rolled-back" if change == "enable" else "restored" if metadata["generation_receipt_sha256"] else "unchanged") + generated = phase in {"active", "restored", "rolled-back"} + receipt = fresh if phase in {"active", "restored"} else intent["wireguard_receipt_sha256"] if phase == "unchanged" else "" + history = guest._Digest(guest._Encode(guest._GenerationAudit(intent, request))) if generated else original + native = {"schema_version": 1, "intent": intent, "request": request, "history_sha256": original, + "admitted_program_sha256": generation["candidate_receipt"]["files"]["lifecycle.py"], "phase": phase, "rollback": not committed} + guest._RuntimeRecord(native, intent) + wg = next(item for item in serviceProof["services"] if item["protocol"] == "wireguard") + + if (generation["source_sha256"] != guest._Digest(guest._Encode(source)) + or metadata["wireguard_receipt_sha256"] != receipt or metadata["generation_receipt_sha256"] != (fresh if generated else "") + or metadata["wireguard_history_sha256"] != history or metadata["native_sha256"] != guest._Digest(guest._Encode(native)) + or wg["receipt_sha256"] != receipt or (wg["state"] == "active") != bool(receipt) + or wg["absence_sha256"] != ("" if receipt else guest._AbsenceDigest(intent["owner"]))): + raise ValueError + + record = authority.Completion({"schema_version": 2, "owner": intent["owner"], "creation_sha256": intent["creation_sha256"], + "operation_id": intent["operation_id"], "intent_sha256": guest._Digest(guest._Encode(metadata)), "parent": metadata["authority_parent"], + "previous_authority_sha256": intent["previous_authority_sha256"], "history_sha256": history, "outcome": value["outcome"]}, intent) + result = {"intent": intent, "request": request, "metadata": metadata, "input": inputProof, "services": serviceProof, + "record": record, "native": native, "values": values} + + except (ValueError, TypeError, KeyError, StopIteration, RecursionError): + pass + + if result is None: + raise CompletionError("Native completion candidate is malformed or differs from its exact reserved terminal witnesses") + + return result + + +def _Files(guest: Any, authority: Any, cache: Any, candidate: dict[str, Any], *, complete: bool) -> dict[str, bytes]: + """Validate every remaining exact sidecar before cleanup; absent sidecars are permitted only after immutable completion exists.""" + + root = authority._Root(guest) + request, intent = candidate["request"], candidate["intent"] + generation = {**cache._Expected(guest, intent, request, guest._ProtocolHistory(intent, request)), "phase": "prepared"} + result = {} + + for filename in ("input.json", "generation.json", "runtime.json"): + if (root / ("." + filename + ".new")).exists() or (root / ("." + filename + ".new")).is_symlink(): + raise CompletionError("Native completion requires canonical terminal sidecars before publication or cleanup") + + path = root / filename + + if not path.exists() and not path.is_symlink(): + if not complete: + raise CompletionError("Native completion is missing its exact terminal sidecars") + + continue + + record = authority._Decode(guest, path) + content = guest._Encode(record) + + if (filename == "input.json" and guest._Digest(content) != candidate["input"]["journal_sha256"] + or filename == "generation.json" and content != guest._Encode(generation) + or filename == "runtime.json" and content != guest._Encode(candidate["native"])): + raise CompletionError("Native completion sidecar differs from its exact verified terminal candidate") + + result[filename] = content + + return result + + +def _WriteCompletion(guest: Any, authority: Any, record: dict[str, Any]) -> None: + """Append only an exact incomplete prefix; a complete immutable completion may only be verified and resynchronized.""" + + root = authority._DurableRoot(guest) + path = root / ("done-" + record["operation_id"] + ".json") + expected = guest._Encode(record) + created = False + + try: + descriptor = os.open(path, os.O_RDWR | os.O_APPEND | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + created = True + + except FileExistsError: + descriptor = os.open(path, os.O_RDWR | os.O_APPEND | os.O_NOFOLLOW | os.O_NONBLOCK) + + try: + if created: + os.fchmod(descriptor, 0o600) + + initial = os.fstat(descriptor) + prefix = authority._CompletionPrefix(guest, record) + linked, current = path.lstat(), os.fstat(descriptor) + + if ((initial.st_dev, initial.st_ino) != (linked.st_dev, linked.st_ino) + or (initial.st_size, initial.st_mtime_ns, initial.st_ctime_ns) != (current.st_size, current.st_mtime_ns, current.st_ctime_ns) + or os.pread(descriptor, len(expected) + 1, 0) != prefix): + raise CompletionError("Native completion inode changed before its exact append") + + offset = len(prefix) + + while offset < len(expected): + written = os.write(descriptor, expected[offset:]) + + if written <= 0: + raise CompletionError("Native completion prefix could not be durably finished") + + offset += written + + os.fsync(descriptor) + + if (os.fstat(descriptor).st_dev, os.fstat(descriptor).st_ino) != (path.lstat().st_dev, path.lstat().st_ino): + raise CompletionError("Native completion inode changed before acknowledgment") + + authority._SyncFile(guest, path, expected) + guest._SyncDirectory(root) + + finally: + os.close(descriptor) + + +def _Retire(guest: Any, authority: Any, filename: str, expected: bytes) -> None: + """Remove only a fixed exact owned sidecar through pinned file/directory identities; never enumerate private runtime payloads.""" + + if filename not in {"input.json", "generation.json", "runtime.json", "pending.json"}: + raise CompletionError("Unsupported native completion cleanup path") + + root = authority._Root(guest) + path = root / filename + with ExitStack() as cleanup: + directory = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + cleanup.callback(os.close, directory) + descriptor = os.open(filename, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) + cleanup.callback(os.close, descriptor) + info = os.fstat(descriptor) + parent = os.fstat(directory) + + def Verify() -> None: + """Reject replacement, hardlinks and namespace rebinding before unlinking the exact known sidecar.""" + + current, linked, linkedParent = os.fstat(descriptor), path.lstat(), root.lstat() + + if (not stat.S_ISREG(current.st_mode) or current.st_uid != guest._ROOT_UID or current.st_nlink != 1 or stat.S_IMODE(current.st_mode) != 0o600 + or (info.st_dev, info.st_ino) != (linked.st_dev, linked.st_ino) + or (info.st_size, info.st_mtime_ns, info.st_ctime_ns) != (current.st_size, current.st_mtime_ns, current.st_ctime_ns) + or not stat.S_ISDIR(linkedParent.st_mode) or linkedParent.st_uid != guest._ROOT_UID or stat.S_IMODE(linkedParent.st_mode) != 0o700 + or (parent.st_dev, parent.st_ino) != (linkedParent.st_dev, linkedParent.st_ino) + or os.pread(descriptor, len(expected) + 1, 0) != expected): + raise CompletionError("Native completion cleanup found a replaced or unexplained sidecar") + + Verify() + os.fsync(descriptor) + Verify() + os.unlink(filename, dir_fd=directory) + os.fsync(directory) + + if (parent.st_dev, parent.st_ino) != (root.lstat().st_dev, root.lstat().st_ino) or path.exists() or path.is_symlink(): + raise CompletionError("Native completion namespace changed during cleanup") + + +def Complete( + guest: Any, peer: Any, authority: Any, policy: Any, inputs: Any, cache: Any, runtime: Any, + value: object, rawIntent: object, *, allowMutation: bool = False, serviceObserver: Any = None, +) -> dict[str, Any]: + """Commit verified terminal guest witnesses once; recorded completion replay is historical metadata recovery, never a health claim.""" + + candidate = Candidate(guest, authority, policy, cache, value, rawIntent) + + if allowMutation is not True: + raise CompletionError("Native completion requires explicit mutation consent") + + intent, record, request = candidate["intent"], candidate["record"], candidate["request"] + pointer = {"name": "done-" + intent["operation_id"] + ".json", "sha256": guest._Digest(guest._Encode(record))} + + with authority._Locks(guest, readOnly=False): + pending = authority._Pending(guest, intent) + head, digest, latest, count = authority._Authority(guest, intent, pending, expectedCompletion=record) + path = authority._Root(guest) / pointer["name"] + recorded = (path.exists() or path.is_symlink()) and authority._CompletionPrefix(guest, record) == guest._Encode(record) + + if pending is None: + if head != pointer or latest != record or not recorded: + raise CompletionError("No exact latest native completion exists for this reservation") + + _WriteCompletion(guest, authority, record) + authority._Publish(guest, "head.json", pointer) + + return dict(record) + + expected = {"schema_version": 1, "intent": intent, "parent": record["parent"], + "deployment_sha256": request["generation"]["cached_receipt"]["deployment_sha256"] if intent["wireguard_state"] == "active" else "", + "phase": "exposed", "fenced": intent["wireguard_state"] == "active"} + + if (guest._Encode(pending) != guest._Encode(expected) or head not in (record["parent"], pointer) + or head == record["parent"] and (digest != intent["previous_authority_sha256"] or count >= authority.MAX_COMPLETIONS) + or (authority._Root(guest) / ".pending.json.new").exists() or (authority._Root(guest) / ".pending.json.new").is_symlink() + or not recorded and ((authority._Root(guest) / ".head.json.new").exists() or (authority._Root(guest) / ".head.json.new").is_symlink())): + raise CompletionError("Native completion does not continue its exact exposed authority or canonical prefix") + + files = _Files(guest, authority, cache, candidate, complete=recorded) + + if not recorded: + with cache.Ready(guest, authority, policy, request, intent, guest._ProtocolHistory(intent, request)) as verify: + native = candidate["native"] + root, unit, _ = guest._Paths(intent["owner"], nonce=request["generation"]["candidate_receipt"]["nonce"]) + installed = {**request["generation"]["candidate_receipt"], "status": "installed"} + active = native["phase"] in {"active", "restored", "unchanged"} + + if native["phase"] == "unchanged": + root, unit, _ = guest._ActivePaths(intent["owner"]) + installed = request["generation"]["cached_receipt"] + + with runtime._Published(guest, root, unit, installed) if active else nullcontext(): + if native["phase"] == "unchanged": + _, history = authority._Observe(guest, peer, intent) + runtime._Active(guest, guest._ActivePaths(intent["owner"])[1]) + + else: + history = runtime._Observe(guest, peer, authority, cache, request, intent, native) + + if history != record["history_sha256"]: + raise CompletionError("Native completion has stale actual consumed history") + + fingerprints = inputs._Compiled(guest, policy, candidate["values"]) + inputRecord = inputs._Load(guest, authority, inputs._Expected(policy, intent, request, fingerprints)) + phase = candidate["input"]["outcome"] + + if inputRecord is None or inputRecord["phase"] != phase: + raise CompletionError("Native completion requires its exact terminal host-input transaction") + + inputs._State(guest, policy, candidate["values"], fingerprints, phase) + observed = inputs._Result(policy, candidate["values"], inputRecord) + + if any(observed[key] != candidate["input"][key] for key in ("journal_sha256", "boot_sha256", "native_sha256")): + raise CompletionError("Native completion host-input witnesses differ from the actual verified policy") + + unrelated = [proof for proof in candidate["services"]["services"] if proof["protocol"] != "wireguard"] + + if unrelated and (serviceObserver is None or guest._Encode(serviceObserver.Observe(guest, candidate["metadata"]["sources"])) != guest._Encode(unrelated)): + raise CompletionError("Native completion requires trusted exact live witnesses for every unrelated service") + + verify() + + if _Files(guest, authority, cache, candidate, complete=False) != files: + raise CompletionError("Native completion witnesses changed before immutable publication") + + _WriteCompletion(guest, authority, record) + verify() + + else: + _WriteCompletion(guest, authority, record) + + if peer._History(guest, intent["owner"]) != record["history_sha256"]: + raise CompletionError("Native completion cannot retire protection with changed or lost peer history") + + authority._Publish(guest, "head.json", pointer) + + for filename, content in files.items(): + if peer._History(guest, intent["owner"]) != record["history_sha256"]: + raise CompletionError("Native completion history changed during sidecar retirement") + + _Retire(guest, authority, filename, content) + + observedHead, _, observedLatest, _ = authority._Authority(guest, intent, pending, expectedCompletion=record) + + if (observedHead != pointer or observedLatest != record or authority._Pending(guest, intent) != pending + or any(path.name.startswith(".") for path in authority._Root(guest).iterdir()) + or any((authority._Root(guest) / filename).exists() or (authority._Root(guest) / filename).is_symlink() + for filename in ("input.json", "generation.json", "runtime.json")) + or peer._History(guest, intent["owner"]) != record["history_sha256"]): + raise CompletionError("Native completion reservation/history changed before retirement") + + _Retire(guest, authority, "pending.json", guest._Encode(pending)) + + return dict(record) diff --git a/src/flayer/profiles/_protocol_generation.py b/src/flayer/profiles/_protocol_generation.py index 9c03c38..8c080a0 100644 --- a/src/flayer/profiles/_protocol_generation.py +++ b/src/flayer/profiles/_protocol_generation.py @@ -50,6 +50,11 @@ def _Reservation( if pending is None or pending["phase"] != "exposed" or pending["parent"] != head or digest != intent["previous_authority_sha256"]: raise GenerationError("Generation preparation requires its exact current exposed reservation") + completion = authority._Root(guest) / ("done-" + intent["operation_id"] + ".json") + + if completion.exists() or completion.is_symlink(): + raise GenerationError("Native completion publication requires finalization before cache preparation") + if request["generation"]["schema_version"] == 3 and runtimeBackend is not None: history = runtimeBackend.Origin(guest, peer, authority, request, intent)["history_sha256"] diff --git a/src/flayer/profiles/_protocol_reservation.py b/src/flayer/profiles/_protocol_reservation.py index bf7e473..2df578a 100644 --- a/src/flayer/profiles/_protocol_reservation.py +++ b/src/flayer/profiles/_protocol_reservation.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import json import os import re @@ -324,7 +325,61 @@ def _Pending(guest: Any, intent: dict[str, Any]) -> dict[str, Any] | None: return values[-1] -def _Authority(guest: Any, intent: dict[str, Any], pending: dict[str, Any] | None) -> tuple[dict[str, str] | None, str, dict[str, Any] | None, int]: +def Completion(value: object, intent: dict[str, Any]) -> dict[str, Any]: + """Validate closed historical cancellation/native completion records without claiming current runtime or cloud health.""" + + intent = Intent(intent) + fields = {"schema_version", "owner", "creation_sha256", "operation_id", "intent_sha256", "parent", "previous_authority_sha256", "history_sha256", "outcome"} + + if (type(value) is not dict or set(value) != fields or type(value["schema_version"]) is not int or value["schema_version"] not in {1, 2} + or not _Hash(value["owner"]) or not _Hash(value["creation_sha256"]) + or value["owner"] != intent["owner"] or value["creation_sha256"] != intent["creation_sha256"] + or type(value["operation_id"]) is not str or _NONCE.fullmatch(value["operation_id"]) is None + or any(not _Hash(value[key]) for key in ("intent_sha256", "previous_authority_sha256", "history_sha256")) + or type(value["outcome"]) is not str or value["outcome"] not in ({"cancelled"} if value["schema_version"] == 1 else {"committed", "not-applied"})): + raise ReservationError("Protocol completion record has invalid ownership or history proof") + + parent = _Pointer(value["parent"]) + + initial = hashlib.sha256(_Json({"domain": "f-layer/ingress/authority/v1", "owner": intent["owner"], "creation_sha256": intent["creation_sha256"]})).hexdigest() + + if value["previous_authority_sha256"] != (parent["sha256"] if parent is not None else initial): + raise ReservationError("Protocol completion does not bind its exact previous authority") + + return dict(value) + + +def _CompletionPrefix(guest: Any, record: dict[str, Any]) -> bytes: + """Observe only this operation's owned exact completion prefix; incomplete bytes grant no completion or mutation authority.""" + + expected = _Json(record) + path = _Root(guest) / ("done-" + record["operation_id"] + ".json") + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + + try: + info = os.fstat(descriptor) + + if (not stat.S_ISREG(info.st_mode) or info.st_uid != guest._ROOT_UID or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) != 0o600 + or not 0 <= info.st_size <= len(expected) or len(expected) > 8192): + raise ReservationError("Protocol completion prefix has unsafe ownership or size") + + content = os.read(descriptor, len(expected) + 1) + current, linked = os.fstat(descriptor), path.lstat() + + if (len(content) != info.st_size or content != expected[:len(content)] + or (info.st_dev, info.st_ino) != (linked.st_dev, linked.st_ino) + or (info.st_size, info.st_mtime_ns, info.st_ctime_ns) != (current.st_size, current.st_mtime_ns, current.st_ctime_ns)): + raise ReservationError("Protocol completion prefix differs from its exact candidate or changed during observation") + + return content + + finally: + os.close(descriptor) + + +def _Authority( + guest: Any, intent: dict[str, Any], pending: dict[str, Any] | None, *, expectedCompletion: dict[str, Any] | None = None, +) -> tuple[dict[str, str] | None, str, dict[str, Any] | None, int]: """Verify every immutable completed parent and reject orphan records rather than reconstructing a lost mutable head.""" root = _Root(guest) @@ -343,20 +398,30 @@ def _Authority(guest: Any, intent: dict[str, Any], pending: dict[str, Any] | Non visited: set[str] = set() latest = None initial = _Initial(guest, intent) + expectedName = "done-" + intent["operation_id"] + ".json" + + if expectedCompletion is not None: + expectedCompletion = Completion(expectedCompletion, intent) + + if (expectedCompletion["schema_version"] != 2 or expectedCompletion["operation_id"] != intent["operation_id"] + or expectedCompletion["previous_authority_sha256"] != intent["previous_authority_sha256"]): + raise ReservationError("Native completion candidate does not consume its exact reservation") while current is not None: if current["name"] in visited or len(visited) >= MAX_COMPLETIONS: raise ReservationError("Protocol completion history is cyclic or exceeds its bound") visited.add(current["name"]) - record = _Decode(guest, root / current["name"]) - - if (set(record) != {"schema_version", "owner", "creation_sha256", "operation_id", "intent_sha256", "parent", "previous_authority_sha256", "history_sha256", "outcome"} - or type(record["schema_version"]) is not int or record["schema_version"] != 1 or record["owner"] != intent["owner"] - or record["creation_sha256"] != intent["creation_sha256"] or record["outcome"] != "cancelled" - or type(record["operation_id"]) is not str or _NONCE.fullmatch(record["operation_id"]) is None - or current["name"] != "done-" + record["operation_id"] + ".json" or guest._Digest(_Json(record)) != current["sha256"] - or any(not _Hash(record[key]) for key in ("intent_sha256", "previous_authority_sha256", "history_sha256"))): + if current["name"] == expectedName and expectedCompletion is not None: + if _CompletionPrefix(guest, expectedCompletion) != _Json(expectedCompletion): + raise ReservationError("Published common head cannot refer to an incomplete completion") + + record = expectedCompletion + + else: + record = Completion(_Decode(guest, root / current["name"]), intent) + + if current["name"] != "done-" + record["operation_id"] + ".json" or guest._Digest(_Json(record)) != current["sha256"]: raise ReservationError("Protocol completion record has invalid ownership or history proof") parent = _Pointer(record["parent"]) @@ -376,8 +441,26 @@ def _Authority(guest: Any, intent: dict[str, Any], pending: dict[str, Any] | Non unheaded = {name for name in names if name.startswith("done-")} - visited + native = pending is not None and pending["phase"] == "exposed" and expectedName in names + + if native and pending is not None: + candidate = (expectedCompletion if expectedCompletion is not None else Completion(_Decode(guest, root / expectedName), intent)) + + if (candidate["schema_version"] != 2 or candidate["operation_id"] != intent["operation_id"] + or candidate["parent"] != pending["parent"] or candidate["previous_authority_sha256"] != intent["previous_authority_sha256"]): + raise ReservationError("Native completion does not consume its exact exposed common authority") + + if expectedCompletion is not None: + _CompletionPrefix(guest, expectedCompletion) + + pointer = {"name": expectedName, "sha256": guest._Digest(_Json(candidate))} + + if (head not in (pending["parent"], pointer) + or ".head.json.new" in names and _Pointer(_Decode(guest, root / ".head.json.new")) != pointer): + raise ReservationError("Native completion has an unexplained common head publication") + if ({name for name in names if name.startswith("done-")} - expected - or (unheaded or ".head.json.new" in names) and (pending is None or pending["phase"] != "cancelling")): + or (unheaded or ".head.json.new" in names) and not native and (pending is None or pending["phase"] != "cancelling")): raise ReservationError("Unrecorded protocol completion or interrupted head requires recovery; missing authority is never adopted") return head, head["sha256"] if head is not None else initial, latest, len(visited) @@ -580,5 +663,6 @@ def Status(guest: Any, peer: Any, value: object) -> dict[str, Any]: _, authority, latest, count = _Authority(guest, intent, pending) return {"status": "protocol-authority", "authority_sha256": authority, "history_sha256": peer.HistoryStatus(guest, intent["owner"])["history_sha256"], - "phase": pending["phase"] if pending is not None else "cancelled" if latest is not None else "initial", + "phase": pending["phase"] if pending is not None else "completed" if latest is not None and latest["schema_version"] == 2 + else "cancelled" if latest is not None else "initial", "reservation_sha256": guest._Digest(_Json(pending)) if pending is not None else "", "completion_count": count} diff --git a/src/flayer/profiles/_protocol_runtime.py b/src/flayer/profiles/_protocol_runtime.py index 1aa2a38..3a962bc 100644 --- a/src/flayer/profiles/_protocol_runtime.py +++ b/src/flayer/profiles/_protocol_runtime.py @@ -35,6 +35,11 @@ def _Reservation(guest: Any, authority: Any, intent: dict[str, Any]) -> dict[str if pending is None or pending["phase"] != "exposed" or pending["parent"] != head or digest != intent["previous_authority_sha256"]: raise RuntimeError("Native generation requires its exact current exposed reservation") + completion = authority._Root(guest) / ("done-" + intent["operation_id"] + ".json") + + if completion.exists() or completion.is_symlink(): + raise RuntimeError("Native completion publication requires finalization before further runtime mutation") + return cast(dict[str, Any], pending) diff --git a/src/flayer/profiles/_wireguard_guest.py b/src/flayer/profiles/_wireguard_guest.py index 3f55c40..ad1a972 100644 --- a/src/flayer/profiles/_wireguard_guest.py +++ b/src/flayer/profiles/_wireguard_guest.py @@ -1963,7 +1963,7 @@ def _OpenLock(*, readOnly: bool) -> int: return descriptor -def _ProtocolGate() -> None: +def _ProtocolGate(*, mutate: bool = False) -> None: """Under WG coordination, refuse a common protocol reservation before ordinary maintenance; service callbacks stay separate.""" root = BASE.parent / "ingress" @@ -1974,6 +1974,31 @@ def _ProtocolGate() -> None: if any((root / name).exists() or (root / name).is_symlink() for name in ("pending.json", ".pending.json.new", ".head.json.new")): raise GuestError("Pending protocol reservation requires explicit protocol recovery before WireGuard maintenance") + if mutate: + descriptor = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + + try: + info = os.fstat(descriptor) + linked = root.lstat() + + if (not stat.S_ISDIR(info.st_mode) or info.st_uid != _ROOT_UID or stat.S_IMODE(info.st_mode) != 0o700 + or (info.st_dev, info.st_ino) != (linked.st_dev, linked.st_ino)): + raise GuestError("Common protocol namespace changed before mutation admission") + + # A failed final completion fsync may leave pending visibly absent + # without durable removal. Reestablish absence before any new write. + os.fsync(descriptor) + _SyncDirectory(root.parent) + linked = root.lstat() + + if ((info.st_dev, info.st_ino) != (linked.st_dev, linked.st_ino) + or linked.st_uid != _ROOT_UID or stat.S_IMODE(linked.st_mode) != 0o700 + or any((root / name).exists() or (root / name).is_symlink() for name in ("pending.json", ".pending.json.new", ".head.json.new"))): + raise GuestError("Common protocol protection changed during mutation admission") + + finally: + os.close(descriptor) + def _GenerationCallback(root: Path, unit: Path, nonce: str, *, start: bool) -> None: """Serialize a nonce-bound leaf callback with WG mutations; native children retain its protecting descriptor.""" @@ -1992,7 +2017,7 @@ def _GenerationCallback(root: Path, unit: Path, nonce: str, *, start: bool) -> N protocol = _RuntimeContext(root, unit, nonce, start=start) if protocol is None: - _ProtocolGate() + _ProtocolGate(mutate=True) request = _Request(root)[0] owner = request["owner"] installed, expectedUnit, transaction = _ActivePaths(owner) @@ -2123,7 +2148,7 @@ def Main(argv: list[str] | None = None) -> int: if descriptor is not None: fcntl.flock(descriptor, (fcntl.LOCK_SH if readOnly else fcntl.LOCK_EX) | fcntl.LOCK_NB) - _ProtocolGate() + _ProtocolGate(mutate=not readOnly) if arguments.remove: _Remove(owner, arguments.expectedReceipt) diff --git a/src/flayer/profiles/_wireguard_peer.py b/src/flayer/profiles/_wireguard_peer.py index d794489..c88e95c 100644 --- a/src/flayer/profiles/_wireguard_peer.py +++ b/src/flayer/profiles/_wireguard_peer.py @@ -755,10 +755,10 @@ def _StageFile(guest: Any, path: Path, content: bytes) -> None: stream.write(content) -def _ProtocolGate(guest: Any) -> None: +def _ProtocolGate(guest: Any, *, mutate: bool = False) -> None: """Under the existing WG lock, refuse unresolved common protocol authority before normal peer/install actions.""" - guest._ProtocolGate() + guest._ProtocolGate(mutate=mutate) def Main(argv: list[str] | None = None) -> int: @@ -849,7 +849,7 @@ def Main(argv: list[str] | None = None) -> int: fcntl.flock(descriptor, (fcntl.LOCK_SH if readOnly else fcntl.LOCK_EX) | fcntl.LOCK_NB) if not options.historyStatus: - _ProtocolGate(guest) + _ProtocolGate(guest, mutate=bool(options.allowMutation)) if installing: if HistoryStatus(guest, owner)["history_sha256"] != options.expectedHistory: diff --git a/tests/unit/test_protocol_completion.py b/tests/unit/test_protocol_completion.py new file mode 100644 index 0000000..d002734 --- /dev/null +++ b/tests/unit/test_protocol_completion.py @@ -0,0 +1,398 @@ +"""Publish native common completion through actual POSIX prefixes with intercepted service/network/cloud boundaries.""" + +from __future__ import annotations + +import copy +import json +import os +import stat +from pathlib import Path +from typing import Any + +import pytest +import test_protocol_generation_cache as cacheFixtures +import test_protocol_generation_input as inputFixtures +import test_protocol_runtime as nativeFixtures +from test_host_input_transaction import Delta +from test_protocol_runtime import Execute, Input +from test_vpn_project_v2 import Snapshot +from test_wireguard_peer_lifecycle import Interrupted + +from flayer.profiles import _host_input_guest as inputs +from flayer.profiles import _host_input_policy as policy +from flayer.profiles import _protocol_completion as completion +from flayer.profiles import _protocol_generation as cache +from flayer.profiles import _protocol_reservation as authority +from flayer.profiles import _protocol_runtime as runtime +from flayer.profiles import _wireguard_guest as guest +from flayer.profiles import _wireguard_peer as peers + +pytestmark = pytest.mark.skipif(os.name != "posix", reason="Common completion needs POSIX metadata/locks; every native/network operation is intercepted") + + +def Sources(identity: dict[str, str]) -> list[dict[str, Any]]: + """Use public synthetic source pins chosen before reservation; private source-file pins still come from the real fixture receipts.""" + + return [{"schema_version": 1, "identity": identity, "pin": {"protocol": name, "intent_fingerprint": "1" * 64, + "material_digest": "2" * 64, "server_digest": "3" * 64}, "kind": "initial", "request_name": "", "request_sha256": ""} + for name in ("amneziawg", "vless-reality", "wireguard")] + + +class Services: + """Model an independent trusted non-WG observer; the request cannot choose this dependency or its measured receipts.""" + + def __init__(self, sources: list[dict[str, Any]], proofs: list[dict[str, str]]) -> None: + """Keep observed service facts independent from the mutable candidate envelope.""" + + self.sources, self.proofs, self.calls = copy.deepcopy(sources), copy.deepcopy(proofs), 0 + + def Observe(self, receiver: Any, sources: list[dict[str, Any]]) -> list[dict[str, str]]: + """Validate source selection and return only public measured facts, with no private data or native output.""" + + assert receiver is guest and receiver._Encode(sources) == receiver._Encode(self.sources) + self.calls += 1 + + return copy.deepcopy(self.proofs) + + +def Environment(tmpPath: Path, monkeypatch: pytest.MonkeyPatch, phase: str = "active") -> tuple[Any, Any, Any, dict[str, Any], dict[str, Any], Services]: + """Reach a real terminal generation/input record while cloud evidence remains the separately trusted controller boundary.""" + + sourceRecords = Sources(Delta().Record()["before"]["identity"]) + sourceSha = guest._Digest(guest._Encode(sourceRecords[-1])) + + def Prepared(*args: Any, **kwargs: Any) -> Any: + """Bind the exact public WG source in the full composite before Reserve/Expose, never by editing reserved authority.""" + + return cacheFixtures.Prepared(*args, **kwargs, sourceSha256=sourceSha) + + monkeypatch.setattr(inputFixtures, "Prepared", Prepared) + change = "enable" if phase in {"active", "rolled-back"} else "disable" + model, network, controller, request, intent = nativeFixtures.Environment(tmpPath, monkeypatch, change) + + if phase != "unchanged": + Execute(controller, request, intent) + + if phase == "active": + Input(request, intent) + + elif phase in {"rolled-back", "restored", "unchanged"}: + if phase == "rolled-back": + Input(request, intent, rollback=True) + + Execute(controller, request, intent, rollback=True) + + if phase != "rolled-back": + Input(request, intent, rollback=True) + + observed = inputs.Status(guest, peers, authority, policy, request, intent, generationCache=cache, runtimeBackend=runtime) + native = authority._Decode(guest, authority._Root(guest) / "runtime.json") + assert native["phase"] == phase + committed = phase in {"active", "inactive"} + values = policy.Transition(request["input"]) + selected = values[1 if committed else 2] + active = [item["transport"] for item in selected["listeners"]] + installed = guest._Digest(guest._Encode({**request["generation"]["candidate_receipt"], "status": "installed"})) + receipt = installed if phase in {"active", "restored"} else intent["wireguard_receipt_sha256"] if phase == "unchanged" else "" + history = peers._History(guest, intent["owner"]) + proof = {"schema_version": 1, "transition": request["input"], "outcome": "applied" if committed else "rolled-back", + **{key: observed[key] for key in ("native_sha256", "journal_sha256", "boot_sha256")}} + services = [{"protocol": source["pin"]["protocol"], "source_sha256": source["pin"]["server_digest"], + "state": "active" if source["pin"]["protocol"] in active else "absent", + "receipt_sha256": receipt if source["pin"]["protocol"] == "wireguard" else "c" * 64, + "absence_sha256": "" if source["pin"]["protocol"] in active else guest._AbsenceDigest(intent["owner"])} for source in sourceRecords] + serviceProof = {"schema_version": 1, "services": services} + configured = {item["transport"]: item for item in [*values[0]["listeners"], request["input"]["listener"]]} + metadata = {"schema_version": 2, "identity": values[0]["identity"], "creation_sha256": intent["creation_sha256"], "parent": None, + "closed_peer_head": None, "closed_peer_head_sha256": intent["creation_sha256"], "closed_peer_count": 0, + "previous_state_sha256": intent["previous_state_sha256"], "sources": sourceRecords, + "configured_listeners": [configured[source["pin"]["protocol"]] for source in sourceRecords], + "active_protocols": [source["pin"]["protocol"] for source in sourceRecords if source["pin"]["protocol"] in active], + "static_fingerprint": "d" * 64, "wireguard_receipt_sha256": receipt, "wireguard_history_sha256": history, + "generation_receipt_sha256": installed if phase in {"active", "restored", "rolled-back"} else "", + "reservation": intent, "authority_parent": None, "ingress_intents": [], "cloud_sha256": "a" * 64, + "input_sha256": guest._Digest(guest._Encode(proof)), "services_sha256": guest._Digest(guest._Encode(serviceProof)), + "generation": request["generation"], "native_sha256": guest._Digest(guest._Encode(native))} + value = {"schema_version": 1, "metadata": metadata, "request": request, "input": proof, "services": serviceProof, + "outcome": "committed" if committed else "not-applied"} + observer = Services(sourceRecords, [item for item in services if item["protocol"] != "wireguard"]) + + return model, network, controller, value, intent, observer + + +def Complete(value: dict[str, Any], intent: dict[str, Any], observer: Services | None) -> dict[str, Any]: + """Exercise the real completion module, metadata files, locks and independent observer dependency.""" + + return completion.Complete(guest, peers, authority, policy, inputs, cache, runtime, value, intent, allowMutation=True, serviceObserver=observer) + + +@pytest.mark.parametrize("phase", ["active", "inactive", "restored", "unchanged", "rolled-back"]) +def test_ExactCompletionCommitsOnceRetiresOnlySidecarsAndReplaysAsHistoricalMetadata(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, phase: str) -> None: + """No completion command changes service/kernel/configuration bytes; replay does not manufacture a live health claim.""" + + model, network, controller, value, intent, observer = Environment(tmp_path, monkeypatch, phase) + before = Snapshot(model.root) if model.root.exists() else None + unrelated, effects = (model.unrelated, network.unrelated), tuple(controller.effects) + expected = completion.Candidate(guest, authority, policy, cache, value, intent)["record"] + result = Complete(value, intent, observer) + assert result == expected and observer.calls == 1 + root = authority._Root(guest) + assert {path.name for path in root.iterdir()} == {"head.json", "done-" + intent["operation_id"] + ".json"} + assert json.loads((root / "head.json").read_bytes()) == {"name": "done-" + intent["operation_id"] + ".json", "sha256": guest._Digest(guest._Encode(result))} + assert (Snapshot(model.root) if model.root.exists() else None) == before + assert (model.unrelated, network.unrelated) == unrelated and tuple(controller.effects) == effects + assert authority.Status(guest, peers, intent)["phase"] == "completed" + assert Complete(value, intent, None) == result and observer.calls == 1 + assert tuple(controller.effects) == effects + + +@pytest.mark.parametrize("boundary", ["done", "head", "input.json", "generation.json", "runtime.json", "pending.json"]) +def test_RecordedCompletionRecoversEveryPublicationAndCleanupBoundaryWithoutRecheckingLiveHealth( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, boundary: str, +) -> None: + """A durable completion is historical evidence: metadata recovery remains possible after a service stops or a reboot blocks its callback.""" + + model, _, controller, value, intent, observer = Environment(tmp_path, monkeypatch) + write, publish, retire = completion._WriteCompletion, authority._Publish, completion._Retire + + def Write(*args: Any) -> None: + """Die after durable immutable completion but before pointer publication.""" + + write(*args) + raise Interrupted + + def Publish(receiver: Any, filename: str, record: dict[str, Any], **kwargs: Any) -> None: + """Die after the exact new common pointer is durable.""" + + publish(receiver, filename, record, **kwargs) + + if filename == "head.json": + raise Interrupted + + def Retire(receiver: Any, backend: Any, filename: str, content: bytes) -> None: + """Die after each individual known sidecar or the final reservation has been retired.""" + + retire(receiver, backend, filename, content) + + if filename == boundary: + raise Interrupted + + with monkeypatch.context() as crash: + if boundary == "done": + crash.setattr(completion, "_WriteCompletion", Write) + + elif boundary == "head": + crash.setattr(authority, "_Publish", Publish) + + else: + crash.setattr(completion, "_Retire", Retire) + + with pytest.raises(Interrupted): + Complete(value, intent, observer) + + model.interface, model.addresses, model.network, model.active = None, [], False, False + controller.active[controller.newUnit.name] = False + calls, effects = observer.calls, tuple(controller.effects) + result = Complete(value, intent, None) + assert result["outcome"] == "committed" and observer.calls == calls + assert tuple(controller.effects) == effects and model.interface is None and not model.network + assert not (authority._Root(guest) / "pending.json").exists() + + +@pytest.mark.parametrize("length", [0, 17, 300]) +def test_OnlyAnExactIncompleteCompletionPrefixCanBeFinishedAfterFreshTerminalVerification( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, length: int, +) -> None: + """The partial record blocks ordinary mutation; completion recovery appends no replacement bytes and reattests the live witnesses.""" + + _, _, controller, value, intent, observer = Environment(tmp_path, monkeypatch) + record = completion.Candidate(guest, authority, policy, cache, value, intent)["record"] + path = authority._Root(guest) / ("done-" + intent["operation_id"] + ".json") + path.write_bytes(guest._Encode(record)[:length]) + path.chmod(0o600) + before = Snapshot(tmp_path) + + with pytest.raises(ValueError): + Execute(controller, value["request"], intent) + + assert Snapshot(tmp_path) == before + assert Complete(value, intent, observer) == record and observer.calls == 1 + assert path.read_bytes() == guest._Encode(record) + + +@pytest.mark.parametrize("fault", ["observer-missing", "observer-drift", "input-proof", "sidecar", "native-temp", "prefix", "peer-history"]) +def test_UnknownOrReboundEvidenceCannotCommitOrRetireAnyProtection(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fault: str) -> None: + """Keep every original byte when attestation is absent or a public envelope differs from its actual terminal journals.""" + + _, _, controller, value, intent, observer = Environment(tmp_path, monkeypatch) + selected: Services | None = observer + root = authority._Root(guest) + + if fault == "observer-missing": + selected = None + + elif fault == "observer-drift": + observer.proofs[0]["receipt_sha256"] = "0" * 64 + + elif fault == "input-proof": + value["input"]["native_sha256"] = "0" * 64 + value["metadata"]["input_sha256"] = guest._Digest(guest._Encode(value["input"])) + + elif fault == "sidecar": + record = authority._Decode(guest, root / "input.json") + record["unknown"] = "preserve unknown metadata" + (root / "input.json").write_bytes(guest._Encode(record)) + + elif fault == "native-temp": + (root / ".runtime.json.new").write_bytes((root / "runtime.json").read_bytes()) + (root / ".runtime.json.new").chmod(0o600) + + elif fault == "prefix": + path = root / ("done-" + intent["operation_id"] + ".json") + path.write_bytes(b"preserve unrelated prefix") + path.chmod(0o600) + + else: + monkeypatch.setattr(peers, "_History", lambda *args: "0" * 64) + + before, effects = Snapshot(tmp_path), tuple(controller.effects) + + with pytest.raises(ValueError): + Complete(value, intent, selected) + + assert Snapshot(tmp_path) == before and tuple(controller.effects) == effects + assert (root / "pending.json").exists() + + +def test_ACommittedNativeCompletionCanParentALaterHistoricalCancellationWithoutReusingItsNonce(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """The original schema-1 cancellation format remains valid after schema-2 completion in the same immutable authority chain.""" + + _, _, _, value, intent, observer = Environment(tmp_path, monkeypatch) + record = Complete(value, intent, observer) + nextIntent = {**intent, "operation_id": "2" * 32, "previous_authority_sha256": guest._Digest(guest._Encode(record)), + "wireguard_state": "active", "wireguard_receipt_sha256": value["metadata"]["wireguard_receipt_sha256"], + "wireguard_absence_sha256": "", "wireguard_history_sha256": record["history_sha256"]} + reserved = authority.Reserve(guest, peers, nextIntent, allowMutation=True) + assert reserved["status"] == "reserved" + assert authority.Cancel(guest, peers, nextIntent, allowMutation=True)["status"] == "cancelled" + assert authority.Status(guest, peers, nextIntent)["completion_count"] == 2 + + with pytest.raises(ValueError): + authority.Reserve(guest, peers, intent, allowMutation=True) + + +@pytest.mark.parametrize("entry", ["guest", "peer"]) +def test_FailedPendingRemovalCannotAdmitAnotherMutationUntilItsNamespaceAbsenceIsDurable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], entry: str, +) -> None: + """Exercise the actual failed final fsync window, then refuse ordinary maintenance/peer admission before any new write.""" + + model, _, _, value, intent, observer = Environment(tmp_path, monkeypatch, "unchanged") + root = authority._Root(guest) + identity = root.stat() + fsync = os.fsync + + def Fail(descriptor: int) -> None: + """Leave a visible but not acknowledged reservation removal; fail the same required barrier on subsequent admission.""" + + info = os.fstat(descriptor) + + if (stat.S_ISDIR(info.st_mode) and (info.st_dev, info.st_ino) == (identity.st_dev, identity.st_ino) + and not (root / "pending.json").exists()): + raise OSError("Synthetic common-directory synchronization failure") + + fsync(descriptor) + + writes = [] + + def Remove(owner: str, receipt: str) -> None: + """Observe attempted ordinary runtime mutation without issuing any native command.""" + + writes.append((owner, receipt)) + + with monkeypatch.context() as fault: + fault.setattr(os, "fsync", Fail) + + with pytest.raises(OSError): + Complete(value, intent, observer) + + assert not (root / "pending.json").exists() and (root / "head.json").exists() + before = Snapshot(tmp_path) + + if entry == "guest": + fault.setattr(guest, "_Remove", Remove) + assert guest.Main(["--remove", "--allow-mutation", "--expected-receipt-sha256", value["metadata"]["wireguard_receipt_sha256"]]) == 1 + capsys.readouterr() + + else: + with authority._Locks(guest, readOnly=False): + with pytest.raises(OSError): + peers._ProtocolGate(guest, mutate=True) + writes.append("peer write must remain unreachable") + + assert not writes and Snapshot(tmp_path) == before + + synchronized = [] + + def Sync(descriptor: int) -> None: + """Record a successful common-directory barrier before admission is allowed to return.""" + + info = os.fstat(descriptor) + + if stat.S_ISDIR(info.st_mode) and (info.st_dev, info.st_ino) == (identity.st_dev, identity.st_ino): + synchronized.append(True) + + fsync(descriptor) + + monkeypatch.setattr(os, "fsync", Sync) + + with authority._Locks(guest, readOnly=False): + (guest._ProtocolGate if entry == "guest" else lambda **kwargs: peers._ProtocolGate(guest, **kwargs))(mutate=True) + assert synchronized + + assert model.root.exists() and not (root / "pending.json").exists() + + +@pytest.mark.parametrize("filename", ["input.json", "generation.json", "runtime.json"]) +def test_ReappearingCanonicalSidecarKeepsTheReservationUntilExplicitRecovery( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, filename: str, +) -> None: + """A known name is insufficient: no sidecar may remain when the final reservation is retired.""" + + _, _, controller, value, intent, observer = Environment(tmp_path, monkeypatch) + root = authority._Root(guest) + content = (root / filename).read_bytes() + retire = completion._Retire + effects = tuple(controller.effects) + + def Retire(receiver: Any, currentAuthority: Any, currentName: str, expected: bytes) -> None: + """Recreate a previously removed canonical sidecar at the final sidecar directory barrier.""" + + retire(receiver, currentAuthority, currentName, expected) + + if currentName == "runtime.json": + (root / filename).write_bytes(content) + (root / filename).chmod(0o600) + + with monkeypatch.context() as fault: + fault.setattr(completion, "_Retire", Retire) + + with pytest.raises(ValueError, match="before retirement"): + Complete(value, intent, observer) + + assert (root / "pending.json").exists() and (root / filename).read_bytes() == content + assert tuple(controller.effects) == effects + assert Complete(value, intent, None)["outcome"] == "committed" + assert not (root / "pending.json").exists() and not (root / filename).exists() + + +def test_ReadOnlyAdmissionDoesNotSynchronizeOrRepairACompletedNamespace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Only a subsequent write grants itself the durable-absence barrier; status keeps its no-repair contract.""" + + _, _, _, value, intent, observer = Environment(tmp_path, monkeypatch) + Complete(value, intent, observer) + before = Snapshot(tmp_path) + monkeypatch.setattr(os, "fsync", lambda *args: pytest.fail("Read-only admission must not synchronize metadata")) + guest._ProtocolGate() + peers._ProtocolGate(guest) + assert Snapshot(tmp_path) == before diff --git a/tests/unit/test_protocol_completion_codec.py b/tests/unit/test_protocol_completion_codec.py new file mode 100644 index 0000000..4aec1c5 --- /dev/null +++ b/tests/unit/test_protocol_completion_codec.py @@ -0,0 +1,156 @@ +"""Cross-bind only public terminal completion data without touching guest files, native commands or cloud APIs.""" + +from __future__ import annotations + +import copy +from dataclasses import replace +from types import SimpleNamespace +from typing import Any + +import pytest +import test_vpn_native_epochs as fixtures + +from flayer import _vpn_protocol_epochs as epochs +from flayer.profiles import _host_input_policy as policy +from flayer.profiles import _protocol_completion as completion +from flayer.profiles import _protocol_generation as cache +from flayer.profiles import _protocol_reservation as authority +from flayer.profiles import _wireguard_guest as guest + + +def Envelope(phase: str = "active") -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: + """Give the pure epoch fixture a real initial authority digest, retaining its exact native/composite/input/service bindings.""" + + epoch = fixtures.Enabled(phase=phase) if phase in {"active", "rolled-back"} else fixtures.Disabled(phase=phase) + metadata = epochs._Decode(epoch.metadata) + intent = metadata["reservation"] + intent["previous_authority_sha256"] = authority._Initial(guest, intent) + request = {"schema_version": 2, "cloud_plan_sha256": epoch.cloud.plan.Fingerprint(), "input": epoch.input.transition.Record(), "generation": metadata["generation"]} + native = {"schema_version": 1, "intent": intent, "request": request, "history_sha256": guest._ProtocolHistory(intent, request), + "admitted_program_sha256": request["generation"]["candidate_receipt"]["files"]["lifecycle.py"], + "phase": phase, "rollback": epoch.cloud.outcome == "not-applied"} + metadata["native_sha256"] = guest._Digest(guest._Encode(native)) + record = {**epochs._Decode(epoch.completion), "previous_authority_sha256": intent["previous_authority_sha256"], + "intent_sha256": guest._Digest(guest._Encode(metadata))} + epoch = replace(epoch, metadata=guest._Encode(metadata), completion=guest._Encode(record)) + value = {"schema_version": 1, "metadata": metadata, "request": request, "input": epoch.input.Record(), + "services": {"schema_version": 1, "services": [proof.Record() for proof in epoch.services]}, "outcome": epoch.cloud.outcome} + + return value, intent, record + + +@pytest.mark.parametrize("phase", ["active", "inactive", "restored", "unchanged", "rolled-back"]) +def test_NativeCompletionReconstructsTheExactRecordAcceptedByTheImmutableEpochReader(phase: str) -> None: + """The guest record has no circular bundle digest and claims only the terminal witnesses selected by its reserved plan.""" + + value, intent, expected = Envelope(phase) + result = completion.Candidate(guest, authority, policy, cache, value, intent) + assert result["record"] == expected and authority.Completion(expected, intent) == expected + + +@pytest.mark.parametrize("fault", ["extra", "alias", "old-generation", "source", "input", "services", "native", "history", "receipt", "parent", "peer", "selection", "cloud", "input-type"]) +def test_CompletionRefusesReboundOrIncompleteWitnessesBeforeAnyGuestAccess(monkeypatch: pytest.MonkeyPatch, fault: str) -> None: + """Rehash modified witness envelopes too; no outer hash can authorize another source, plan, phase or receipt.""" + + value, intent, _ = Envelope("restored") + value = copy.deepcopy(value) + metadata = value["metadata"] + + if fault == "extra": + value["private_key"] = "suppressed-canary" + + elif fault == "alias": + metadata["schema_version"] = True + + elif fault == "old-generation": + value["request"]["generation"]["schema_version"] = 2 + del value["request"]["generation"]["callback_abi"] + intent["plan_sha256"] = guest._Digest(guest._Encode(value["request"])) + + elif fault == "source": + metadata["sources"][-1]["pin"]["server_digest"] = "0" * 64 + + elif fault == "input": + value["input"]["boot_sha256"] = "0" * 64 + + elif fault == "input-type": + value["input"]["schema_version"] = True + + elif fault == "services": + value["services"]["services"][-1]["source_sha256"] = "0" * 64 + + elif fault == "native": + metadata["native_sha256"] = "0" * 64 + + elif fault == "history": + metadata["wireguard_history_sha256"] = intent["wireguard_history_sha256"] + + elif fault == "receipt": + metadata["wireguard_receipt_sha256"] = intent["wireguard_receipt_sha256"] + + elif fault == "parent": + metadata["authority_parent"] = {"name": "done-" + "a" * 32 + ".json", "sha256": "0" * 64} + + elif fault == "peer": + metadata["closed_peer_count"] = True + + elif fault == "selection": + metadata["active_protocols"] = ["wireguard"] + + else: + metadata["cloud_sha256"] = [] + + metadata["input_sha256"] = guest._Digest(guest._Encode(value["input"])) + metadata["services_sha256"] = guest._Digest(guest._Encode(value["services"])) + monkeypatch.setattr(authority, "_Root", lambda *args: pytest.fail("Refused completion data must not inspect guest files")) + monkeypatch.setattr(guest, "_Run", lambda *args, **kwargs: pytest.fail("Refused completion data must not execute a native command")) + + with pytest.raises(completion.CompletionError): + completion.Complete(guest, SimpleNamespace(), authority, policy, SimpleNamespace(), cache, SimpleNamespace(), value, intent, allowMutation=True) + + +def test_WellFormedCompletionStillRequiresExplicitMutationConsentBeforeAnyGuestAccess(monkeypatch: pytest.MonkeyPatch) -> None: + """A complete data-only candidate does not itself authorize publishing common authority.""" + + value, intent, _ = Envelope() + monkeypatch.setattr(authority, "_Root", lambda *args: pytest.fail("Missing consent must not inspect guest files")) + + with pytest.raises(completion.CompletionError, match="consent"): + completion.Complete(guest, SimpleNamespace(), authority, policy, SimpleNamespace(), cache, SimpleNamespace(), value, intent) + + +@pytest.mark.parametrize("fault", ["schema", "outcome", "owner", "creation", "operation", "parent", "previous", "history", "extra"]) +def test_CommonCompletionCodecKeepsHistoricalAndNativeOutcomesClosed(fault: str) -> None: + """Existing cancellations remain schema 1; committed/not-applied native records use an explicit different schema.""" + + _, intent, record = Envelope() + + if fault == "schema": + record["schema_version"] = True + + elif fault == "outcome": + record["outcome"] = "cancelled" + + elif fault == "owner": + record["owner"] = "0" * 64 + + elif fault == "creation": + record["creation_sha256"] = "0" * 64 + + elif fault == "operation": + record["operation_id"] = "../elsewhere" + + elif fault == "parent": + record["parent"] = {"name": "../foreign", "sha256": "0" * 64} + + elif fault == "previous": + record["previous_authority_sha256"] = "0" * 64 + + elif fault == "history": + record["history_sha256"] = False + + else: + record["unknown"] = "suppressed-canary" + + with pytest.raises(ValueError): + authority.Completion(record, intent) diff --git a/tests/unit/test_protocol_generation_cache.py b/tests/unit/test_protocol_generation_cache.py index b6a6356..a9871d9 100644 --- a/tests/unit/test_protocol_generation_cache.py +++ b/tests/unit/test_protocol_generation_cache.py @@ -26,6 +26,7 @@ def Prepared( tmpPath: Path, monkeypatch: pytest.MonkeyPatch, change: str = "disable", version: int = 1, helper: bytes | None = None, + *, sourceSha256: str = "f" * 64, ) -> tuple[Any, dict[str, Any], dict[str, Any], Path, Path | None]: """Expose one exact generation plan while retaining independent old runtime and source snapshots.""" @@ -58,7 +59,7 @@ def Prepared( intent.update(wireguard_state="absent", wireguard_receipt_sha256="", wireguard_absence_sha256=guest._AbsenceDigest(model.owner)) value = {"schema_version": 2, "cloud_plan_sha256": "a" * 64, "input": transition.Record(), - "generation": {"schema_version": version, "change": change, "source_sha256": "f" * 64, + "generation": {"schema_version": version, "change": change, "source_sha256": sourceSha256, "source_files": {name: receipt["files"][name] for name in generations._SOURCE_FILES}, "cached_receipt": receipt, "candidate_receipt": desired}} diff --git a/tests/unit/test_protocol_reservation.py b/tests/unit/test_protocol_reservation.py index 509846d..b7ee563 100644 --- a/tests/unit/test_protocol_reservation.py +++ b/tests/unit/test_protocol_reservation.py @@ -200,11 +200,11 @@ def test_NormalPeerAndMaintenanceEntrypointsRefusePendingAuthorityButAllowHistor originalGate = peers._ProtocolGate checked = [] - def Check(current: Any) -> None: + def Check(current: Any, *, mutate: bool = False) -> None: """Prove that failure comes from the intended gate, retaining the entrypoint's fixed sanitized diagnostic.""" checked.append(True) - originalGate(current) + originalGate(current, mutate=mutate) monkeypatch.setattr(peers, "_ProtocolGate", Check)