diff --git a/AGENTS.md b/AGENTS.md index 8e672f0..2552251 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,6 +12,8 @@ ## 批量节点接入：默认启动一个子助手 +后续新批次默认使用 [正式适配器入口](docs/maintenance/NODE_BATCH_RUNTIME.md)，由 `configure-site.py` 固定计划，再由 `batch.py` 的 `ssh-site-v1` 运行；动态读取验收账号应得节点集合，不固定节点数量。没有有效验收账号时自动创建本任务受限测试用户并负责禁用收尾；SSH始终经已登记白色机。具体实现、证据续跑和原服务调查边界见该专题；旧冻结计划及其处理脚本不改摘要迁就新实现。仅维护脚本/文档不触发真实接入或线上创建测试用户。 + 用户要求执行批量节点接入（明确说批量，或同批接入两个及以上节点）时，主协调者必须默认启动一个子助手，无需再次询问是否使用助手：核查验收助手负责候选服务器、端口、中转与已有资源冲突核查，并独立审查脚本、测试并发/查重/恢复及执行接入验收。主协调者负责统一参数、展示编号、端口、唯一账本及生产写入。完整分工、执行关卡与消耗控制见 [批量接入协作流程](docs/maintenance/NODE_BATCH.md#多智能体协作流程)。 这项授权针对当前任务的子助手，不自动新建或分叉普通聊天，也不授权向已有聊天发消息。仅讨论方案、维护脚本/文档或单节点任务不自动触发。若子助手工具不可用、并发名额不足或助手失败，说明具体缺口并恢复/等待；未经用户明确允许，不静默省略角色或跳过验收。当前批次已有对应助手时复用，不重复启动。具体线上变更仍以用户本次接入授权为限；项目规则不启动无人值守接入。 diff --git a/docs/maintenance/NODE_BATCH.md b/docs/maintenance/NODE_BATCH.md index f71a47c..bad216c 100644 --- a/docs/maintenance/NODE_BATCH.md +++ b/docs/maintenance/NODE_BATCH.md @@ -6,6 +6,8 @@ ## 文件与运行方式 +2026-10-17新增[正式站点适配器](NODE_BATCH_RUNTIME.md)：后续新批次默认用`configure-site.py`生成`ssh-site-v1`计划，具备动态完整节点验收、受限测试账号、白色机SSH会话复用、证据续跑及阶段事件。已通过本地隔离测试，尚未部署或线上演练。本页下述外部命令适配器协议保留兼容，旧任务脚本和既有冻结摘要不修改；旧命令适配器不透传缓存证据，继续逐阶段现场probe。 + | 文件 | 用途 | | --- | --- | | `scripts/node-onboarding/batch.py` | 参数校验、资源预留、并发、进程间锁、逐阶段账本与恢复 | diff --git a/docs/maintenance/NODE_BATCH_RUNTIME.md b/docs/maintenance/NODE_BATCH_RUNTIME.md new file mode 100644 index 0000000..90f555b --- /dev/null +++ b/docs/maintenance/NODE_BATCH_RUNTIME.md @@ -0,0 +1,111 @@ +# 批量接入正式适配器 + +本专题是后续批量接入的默认技术入口。适配器实现、隔离测试及文档于2026-10-17完成；**本轮没有新增线上节点、部署新运维版本或运行真实SSH/客户端验收**。上一轮三节点实际验收使用的是已冻结旧任务适配器，不能套用为新实现的生产验收。 + +`batch.py`保留原命令适配器兼容，新计划默认用`configure-site.py`生成`runtime.kind=ssh-site-v1`。旧计划、账本、任务handler一律保留，不修改源码摘要后直接续跑旧任务。生产操作仍须具体接入授权、独立助手审核、Git运维部署及现场验收。 + +## 文件与职责 + +| 文件 | 职责 | +| --- | --- | +| `configure-site.py` | 本地合并明确站点参数，固定解释器、Paramiko版本、源码与非秘密配置摘要；不联网、不写账本、不覆盖旧计划 | +| `site_adapter.py` | 复用正式stage/provision/promote流程；按国家、组、批次和路径参数执行，按需连接服务器，进行动态正式订阅验收 | +| `ssh_pool.py` | 线程内复用白色机及下游SSH会话；下游强制通过白色机direct-tcpip，不提供直连回退 | +| `panel-acceptance.php` | 开工前选择有效正式组账号；缺少时创建本批受限非管理员测试用户，记录归属并收尾禁用 | +| `acceptance.py` | 动态集合、逐节点结果、原始证据摘要、QA身份/禁用/拒绝/下发排除和时效的完整自动审计 | + +## 准备与Git部署 + +先按CoDocs确认VPS、白色机访问路线、正式组、展示编号及资源冲突。Python执行环境使用`requirements.txt`的Paramiko 5.0.0；配置与执行必须使用同一解释器。安装来源限官方包源/已审核环境，不临时导入任务目录中的未知transport模块。 + +生产前先提交并推送源码，再生成、推送和部署独立运维Git版本：保留原安装脚本，加入`panel-acceptance.php`及其`operations.php`，还有stage/promote/rollback/readiness/verify-node入口；面板、网关和目标节点的对应运维目录必须为计划中完整`provision_commit`且无工作区差异。配置文件和凭据不入Git。独立快照由正式`bundle-ops.py`生成，保持基础运维树、工作目录和真实Git索引不变，仅复制10个明确的已提交运维文件，拒绝覆盖已有分支或bundle，不联网： + +```sh +python3 scripts/node-onboarding/bundle-ops.py --source FULL_SOURCE_COMMIT --base FULL_OPS_BASE_COMMIT --branch codex/node-tools-BATCH --bundle /private/task/provision.bundle +``` + +随后按已授权任务推送返回的运维分支、审核commit/bundle摘要，再经白色机部署到明确版本。当前源码/运维快照的提交推送与远端部署分别报告；本轮不自动切换任何生产应用或旧任务的运维目录。 + +适配器在stage前检查本地依赖、远端Git版本及工具、正式组和账号、节点既有配置、独立端口、原服务、证书归属与钩子。既有证书无本任务owner、全局证书钩子或目标证书的额外pre/post钩子均停止。缺依赖须在准备阶段解决，不先创建一批面板对象后再临时安装工具。 + +## 本地参数文件 + +全部放在权限700的任务目录，文件600。基础计划沿用[NODE_BATCH](NODE_BATCH.md#参数格式)，包括本批任意数量、任意已支持国家的候选节点；不固定五节点、日本、组1或任务日期。新增站点配置JSON如下，示例路径必须换为实际已核实值： + +```json +{ + "ops_root": "/opt/dpl-batch/provision", + "node_root": "/opt/dpl-node/provision", + "batch_backup_root": "/root/codex-backups/dpl/BATCH", + "node_backup_root": "/root/codex-backups/dpl/BATCH-nodes", + "qa_root": "/var/lib/dpl-sgw/BATCH", + "provider_snapshot": "/private/task/providers.json", + "exit_registry": "/private/task/exits.json", + "ssh_config": "/private/task/ssh.json", + "preserved_service": { + "unit": "XrayR", "binary": "/usr/local/XrayR/XrayR", + "config": "/etc/XrayR/config.yml", "port": 443 + }, + "formal_socks_port": 19203, + "client_sha256": "填已核验官方客户端文件的64位SHA256", + "gateway_containers": ["dpl-subscribe-gateway", "dpl-subscribe-admin", "subscribe-gateway", "subscribe-admin"] +} +``` + +批次备份目录必须独立且固定，不跨实际批次复用旧资源基线。新节点端口、正式客户端端口均查重预留；共享网关客户端还使用同一账本内的进程锁。当前适配器用于独立AnyTLS节点及明确列出的既有服务；不自动把白色机、面板或网关当节点宿主。 + +`ssh.json`含`known_hosts`绝对路径及`hosts`映射，key为准确CoDocs VPS ID。每台登记`host`、`port`、`username`、用户已批准的`host_key_sha256`（OpenSSH SHA256格式）、可选`credential_ref`；白色机额外登记`role=white_jump`，ID须与`context.jump_vps`一致。候选IP必须与冻结计划相同。`credential_ref`只引用既定安全存储中的600文件。默认JSON支持password/key_filename/passphrase；既有CoDocs Markdown可设`credential_format=codocs-password`，仅按原文件中唯一的“密码：”字段在内存读取，不能为了适配而复制或改写原凭据正文。内容不得进入计划、日志、argv或Git。配置和出口登记摘要固定，凭据只按需读取，不默认自动轮换。 + +只从known-hosts文件选择与冻结配置中已批准指纹完全相符的公钥；找不到匹配时在读取/发送凭据前停止。全局known-hosts增加无关记录无需改变本批受信任指纹。不关闭主机密钥检查，不自动TOFU；首次信任先按访问规则核验/批准精确指纹，保存公开known-host条目。会话丢失只标记未知，不自动重发安装/开放命令；下次仍从现场probe恢复。 + +`exits.json`按SNI登记独立已知预期，例如： + +```json +{"node.example.invalid":{"expected_exit":"192.0.2.11","country":"JP"}} +``` + +覆盖该账号当前应获得的全部旧节点与新候选，不能从本次实测结果反填预期来让检查通过。增加或改变节点时，按CoDocs/授权现场证据更新新任务的登记；遇到未知预期或不支持协议停止，不能静默省略。当前站点适配器针对IPv4 A记录、固定端口AnyTLS；端口范围或其他协议停止而非静默省略，扩展需另补验证能力。 + +`providers.json`是实际已登记后台UI或授权官方API的近期观察，最多900秒有效，不使用语言相关DOM字符串： + +```json +{ + "captured_at": 0, + "dns": [{"type":"A","name":"node.example.invalid","address":"192.0.2.11","proxy":false}], + "relays": [{"name":"dpl-new-anytls","host":"relay.example.invalid","port":47011,"target":"192.0.2.11","listen_port":24443,"proxy_protocol_in":false,"proxy_protocol_out":false}] +} +``` + +`captured_at`填实际采集Unix时间。观察文件可以刷新，不能把静态勾选当现场证据。资源存在但参数/归属不符停止；尚未创建的DNS/中转仍按注册后台批量操作、更新观察后续跑。本实现没有新增Cloudflare/中转官方API写入权限，也不默认直连其源站。 + +## 执行与等待 + +```sh +/path/to/reviewed/python scripts/node-onboarding/configure-site.py --plan /private/task/base.json --site /private/task/site.json --output /private/task/frozen.json +/path/to/reviewed/python scripts/node-onboarding/batch.py plan --plan /private/task/frozen.json --state-root /private/task/ledger +/path/to/reviewed/python scripts/node-onboarding/batch.py run --plan /private/task/frozen.json --state-root /private/task/ledger --jobs 3 --timeout 900 --execute --events +``` + +配置/plan只做本地工作，不创建线上测试用户。`run --execute`进入准备前检时才允许准备验收账号，随后全批前检屏障通过才stage。`--events`输出无秘密的阶段状态JSON行，最后一行为批次结果；不加此参数保持原单JSON输出协议。账本`events.jsonl`仅追加状态变化，等待时使用进程输出/事件，不每几秒重复查全部状态。只有有新依赖证据、外部DNS/中转变化或用户恢复指令时续跑，不能无限重试。 + +SSH会话仅在本次进程、同一工作线程内复用，线程不共用可变SSH客户端；进程退出关闭连接，不建立后台守护进程。每个阶段通过lazy连接只访问必要主机。Paramiko channel有远端命令超时；进程内模式不承诺`--timeout`能杀死已发送远端命令，停止后仍须核实实际结果。 + +## 动态验收和测试用户 + +准备阶段先按正式组、banned、额度和有效期选既有账号，不把管理员身份自动排除；检查真实`ServerService::getAvailableServers`及`getAvailableUsers`，包括插件过滤。没有可用账号时自动创建本任务测试用户：非管理员、100MiB、1小时、1设备、无订单/套餐改写及通知消息。私有journal在DB提交前记录身份，丢失回包时恢复同一用户，不重复创建或接管同名用户。 + +新节点仍沿用各自隔离QA→严格TLS/出口/上报→独立助手按精确报告审核→append开放并禁用QA。全部本组候选开放后，正式验收按账号**当前实际应获得的集合**生成完整快照，与客户端完整订阅逐条校验，并逐条实际联网。集合绑定节点ID、标签、端点、SNI、独立预期出口/国家；验收后再取一次集合，增加、删除或改变任一项均阻止旧报告通过，不固定节点数。旧节点同样必须覆盖。 + +每组正式客户端串行运行，保留完整订阅DNS/路由，只给诊断目的地固定出站。新建测试用户可能等待节点拉取用户列表；只对通用客户端运行失败最多做3轮诊断读测试，间隔15秒且共用900秒预算。结构、TLS或出口断言错误不重试；不重放生产写入、不另建用户来绕过失败。证据绑定完整计划摘要、节点集合摘要、原始报告及QA拒绝文件SHA256、QA实际ID、403、banned、实际下发排除及900秒时效。任一缺项或摘要/集合变化失败。旧节点关键配置、全部套餐、原用户关联和订阅地址按批次原始基线核对，网关容器ID/启动时间/重启数再次核对。 + +正式验收成功或失败都禁用本任务新建的正式测试用户；既有账号不被禁用或改密码/权限。进程正常退出在finally收尾，即使等待或部分前检失败也尝试清理；遇到断网/SIGKILL无法保证立即清理，记录的归属、1小时有效期与额度是兜底，恢复后先核对旧journal。已关闭用户绝不重新启用；下一次需要用户时创建新受限租约，保留历史并在最终审计核对其403/下发排除。完成要求由服务端实际集合/订阅拒绝证明，不承诺立即踢断已经建立的客户端会话。 + +## 续跑与原服务调查 + +preflight/publish/audit始终现场执行。只有先前状态为done、收到真实前检依赖摘要、请求/代码/config未变、证据未过期，才可复用stage/dns/provision/relay/verify。前驱依赖变化使后续缓存失效；running/unknown/conflict绝不跳过。旧适配器不提供这些证据时仍按原逐阶段probe逻辑执行。生产写入仍先probe、持久化意图、apply后probe，不为速度跳过归属检查。 + +原服务保存最初unit/程序/配置SHA256、监听、active及PID/启动时间/重启数。配置/程序/监听改变或不可用停止；仅进程变化分类为“需调查”，不宣称是DPL导致，也不自动刷新基线。记录观察后由操作者及助手核查生命周期、OOM、操作时间线和稳定采样；认可时在本节点私有备份登记`preserved-service-review.json`，绑定最初/本次观察摘要、原因及证据SHA256。原基线保持；再次重启不匹配旧审核，须重新调查。 + +## 验证范围 + +新增Python测试覆盖动态1/2/7/23节点、集合变化/漏旧节点、出口/TLS/QA身份/摘要/时效、白色机跳转及严格密钥策略、线程会话复用/断线、依赖续跑及未知/过期状态、原服务精确调查回执、远端脚本与PHP构造。PHP隔离测试使用真实Eloquent/SQLite和ServerService，覆盖自动受限用户、既有管理员复用、真实下发、无效账号、journal恢复、关闭后新租约、归属漂移、插件排除及失败收尾；不加载.env/应用provider/Redis/HTTP。生产停机、失败回退、远端真实会话复用性能及十节点耗时尚未演练，不承诺固定提速倍数。 diff --git a/docs/maintenance/NODE_RUNBOOK.md b/docs/maintenance/NODE_RUNBOOK.md index d97b881..a90c38d 100644 --- a/docs/maintenance/NODE_RUNBOOK.md +++ b/docs/maintenance/NODE_RUNBOOK.md @@ -84,6 +84,8 @@ esac ## 脚本能力与边界 +后续新批次按[正式适配器](NODE_BATCH_RUNTIME.md)准备和执行：动态账号授权集合验收、缺账号时自动创建受限用户、白色机SSH复用及证据有效期续跑。旧命令适配器继续兼容。下文运维提交为上一轮三节点已验收版本，新适配器与PHP验收用户入口尚未部署，不能把旧任务验收视为新版本线上通过。 + 2026-10-17：下列脚本已完成 PHP 语法检查、面板隔离测试（9 tests / 65 assertions）及批量进程测试（11 tests），源码提交并推送 `6af6764`；独立运维快照 `b8d10fac963d63825c97d36d41ff7f525116bbfd` 已部署到本次面板/网关的 `/opt/dpl-batch/provision` 与三台新节点 `/opt/dpl-node/provision`。生产应用代码未随之更新。各阶段现场结果以 [STATUS](STATUS.md) 和 [接入记录](NODE_ONBOARDING.md) 为准；隔离测试中的通知替身不证明生产通知成功。 | 工具 | 新行为 | 边界 | diff --git a/docs/maintenance/README.md b/docs/maintenance/README.md index a7e8221..47f23c8 100644 --- a/docs/maintenance/README.md +++ b/docs/maintenance/README.md @@ -24,6 +24,7 @@ | NODE_ONBOARDING.md | 五节点链路、命名规则、共享资源与验收 | | NODE_RUNBOOK.md | initialize/append参数、脚本入口、前检/证书/验收与面板回退；隔离测试/运维版本与现场边界，含验证清单 | | NODE_BATCH.md | 批量接入默认一个子助手的分工与关卡；append并发锁/查重/断点恢复、命令协议与测试边界 | +| NODE_BATCH_RUNTIME.md | 新批次默认正式适配器、动态集合验收、自动受限测试用户、白色机SSH复用、证据续跑及部署准备 | | GATEWAY_ADMIN_PROXY.md | 两网关管理域名、白色机新旧区域反代、独立证书续期、恢复和验收边界 | 这些说明用于辅助 AI 接手，不代表 AI 会在聊天结束后自动持续运行。应用定时任务在服务器执行，AI 维护任务需要会话或明确配置的自动化来启动。 diff --git a/docs/maintenance/STATUS.md b/docs/maintenance/STATUS.md index ddf1c35..c3a4f98 100644 --- a/docs/maintenance/STATUS.md +++ b/docs/maintenance/STATUS.md @@ -2,6 +2,14 @@ 文档核对日期：2026-10-11。以下区分本会话检查结果与用户验收确认，不承诺服务器一直保持此状态。 +## 2026-10-17 批量接入流程优化（本地验证） + +已实现正式`ssh-site-v1`适配器及`configure-site.py`入口，去除临时handler中的固定日期、日本、组1和节点数量。正式验收从真实账号授权集合生成快照，完整校验新旧节点、预期出口、严格TLS、原始证据摘要、QA身份/403/禁用/下发排除及时效；集合变化停止。没有有效正式组账号时自动创建100MiB/1小时/1设备非管理员测试用户，journal恢复同一身份，正常成功/失败/等待均清理本任务用户，关闭后不复活；异常中断依靠日志恢复和时限兜底。已批准指纹匹配后才读取凭据，SSH始终经白色机，按线程复用会话、按阶段延迟连接，不自动重放断线命令。 + +续跑保留全批前检及publish/audit现场关口，以请求/代码/当前依赖摘要和有效期复用已完成步骤，前驱变化向下游失效；事件输出替代重复状态轮询。原服务以unit/程序/配置/监听和进程信息区分实际冲突与待调查重启，原基线不覆盖。独立运维打包工具从已提交源码生成可追踪快照，保留工作区/真实索引及旧运维基础树；旧命令适配器及旧冻结任务文件不修改。 + +本轮Python 30项测试通过；PHP新增9项/35断言及既有9项/65断言隔离测试通过，均未加载真实.env或联网。CoDocs全库离线检查通过，前次.obsidian权限问题已由用户处理，本轮未读写其配置正文。新版本尚未部署至远端或进行真实SSH/客户端性能演练，不能把上一轮三节点验收复用为新实现线上通过，也不承诺十节点耗时。准备/部署/执行与边界见[正式适配器专题](NODE_BATCH_RUNTIME.md)；后续新批次默认按该入口执行。 + ## 2026-10-17 三节点批量接入完成 新增日本JP - 02/03/04（node-jp03/04/05），均已append正式组1、QA禁用。五节点正式完整订阅逐条严格TLS联网、预期IPv4出口/国家及三个QA订阅403通过；旧节点、套餐/用户关联、订阅设置与四个网关容器基线保持。续期演练逐台通过。源码6af6764、独立运维b8d10fac已提交推送并部署运维目录，生产应用代码未随之更新；9项/65断言面板隔离测试及11项批量进程测试通过。 diff --git a/scripts/node-onboarding/acceptance.py b/scripts/node-onboarding/acceptance.py new file mode 100644 index 0000000..c62470a --- /dev/null +++ b/scripts/node-onboarding/acceptance.py @@ -0,0 +1,93 @@ +"""Token-free dynamic evidence validation shared by the reusable site adapter.""" +import hashlib +import ipaddress +import re +import json +import time + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(',', ':')).encode()).hexdigest() + + +def snapshot(nodes): + if not isinstance(nodes, list) or not nodes: + raise ValueError('Empty authorized node set') + fields = ('id', 'tag', 'relay', 'relay_port', 'sni', 'expected_exit', 'country') + result = [] + for node in nodes: + if node.get('type') != 'anytls' or any(node.get(k) in (None, '') for k in fields): + raise ValueError('Unsupported protocol or incomplete independently registered expectations') + if type(node['id']) is not int or node['id']<1 or type(node['relay_port']) is not int or not 1<=node['relay_port']<=65535 or not isinstance(node['country'],str) or not re.fullmatch('[A-Z]{2}',node['country']): + raise ValueError('Invalid authorized node identity or expectation') + address=ipaddress.ip_address(node['expected_exit']) + if getattr(address,'scope_id',None):raise ValueError('Scoped exit expectation unsupported') + result.append({k: node[k] for k in fields}) + for key in ('id', 'tag'): + if len({n[key] for n in result}) != len(result): + raise ValueError('Duplicate authorized node identity') + if len({(n['relay'], n['relay_port'], n['sni']) for n in result}) != len(result): + raise ValueError('Ambiguous authorized endpoint') + return sorted(result, key=lambda n: n['id']) + + +def audit(before, after, report_bytes, qa_bytes, binding, plan_sha, expected_qas, now=None, ttl=900): + now = time.time() if now is None else now + before, after = snapshot(before), snapshot(after) + if before != after or binding.get('snapshot_sha256') != digest(before): + raise ValueError('Authorized node set changed; regenerate and verify the complete set') + if binding.get('plan_sha256') != plan_sha: + raise ValueError('Wrong plan binding') + for label, raw in [('report', report_bytes), ('qa_rejections', qa_bytes)]: + if binding.get(label + '_sha256') != hashlib.sha256(raw).hexdigest(): + raise ValueError('Evidence bytes changed') + report, qas = json.loads(report_bytes), json.loads(qa_bytes) + for stamp in (binding.get('verified_at'), report.get('verified_at'), qas.get('captured_at')): + if type(stamp) not in (int, float) or not 0 <= now-stamp <= ttl: + raise ValueError('Evidence expired or future dated') + if report.get('state') != 'live_verified' or len(report.get('results', [])) != len(before): + raise ValueError('Incomplete live verification') + expected = {(n['tag'], n['relay'], n['relay_port'], n['sni'], n['expected_exit'], n['country']) for n in before} + actual = set() + for n in report['results']: + if n.get('live') is not True or n.get('strict_tls') is not True: + raise ValueError('Live strict TLS required for every node') + actual.add(tuple(n.get(k) for k in ('tag', 'relay', 'relay_port', 'sni', 'exit_ip', 'country'))) + if actual != expected: + raise ValueError('Node/endpoint/exit/country mismatch') + if qas.get('plan_sha256') != plan_sha or set(qas.get('users', {})) != set(expected_qas): + raise ValueError('QA evidence identity mismatch') + for key, identity in expected_qas.items(): + q = qas['users'][key] + if q.get('id') != identity or q.get('http_status') != 403 or q.get('banned') is not True or q.get('excluded') is not True: + raise ValueError('QA account not fully revoked') + return {'nodes': len(before), 'snapshot_sha256': digest(before), 'verified_at': now} + + +def service_change(before, current): + """Process movement is investigatory; never silently accept/rebase it.""" + required = ('unit_sha256', 'binary_sha256', 'config_sha256', 'listeners') + if any(not before.get(k) or not current.get(k) for k in required): + return 'unknown' + if any(before[k] != current[k] for k in required) or current.get('active') is not True: + return 'conflict' + if any(before.get(k) != current.get(k) for k in ('pid', 'started', 'restarts')): + return 'investigate_restart' + return 'ready' + + +def provider_state(observed, node, now=None): + """Canonical observations from the registered provider UI/API, never HTML substrings.""" + now=time.time() if now is None else now + stamp=observed.get('captured_at') + if type(stamp) not in (int,float) or not 0<=now-stamp<=900: + raise ValueError('Provider observation expired') + if not isinstance(observed.get('dns'),list) or not isinstance(observed.get('relays'),list): + raise ValueError('Canonical DNS and relay observations required') + dns=[d for d in observed['dns'] if d.get('name')==node['tls_domain']] + relays=[r for r in observed['relays'] if r.get('host')==node['relay_host'] and r.get('port')==node['relay_port']] + if dns and (len(dns)!=1 or dns[0].get('type')!='A' or dns[0].get('address')!=node['node_address'] or dns[0].get('proxy') is not False): + raise ValueError('DNS provider resource conflict') + if relays and (len(relays)!=1 or relays[0].get('target')!=node['node_address'] or relays[0].get('listen_port')!=node['listen_port'] or relays[0].get('name')!='dpl-'+node['key']+'-anytls' or relays[0].get('proxy_protocol_in') is not False or relays[0].get('proxy_protocol_out') is not False): + raise ValueError('Relay provider resource conflict') + return {'dns':dns,'relays':relays} diff --git a/scripts/node-onboarding/batch.py b/scripts/node-onboarding/batch.py index baeb5de..01d349e 100644 --- a/scripts/node-onboarding/batch.py +++ b/scripts/node-onboarding/batch.py @@ -6,6 +6,7 @@ import contextlib import fcntl import hashlib import ipaddress +import importlib.util import json import os from pathlib import Path @@ -20,6 +21,7 @@ import time PHASES = ('stage', 'dns', 'provision', 'relay', 'verify', 'publish', 'audit') SHARED = {'stage': 'panel', 'dns': 'dns', 'relay': 'relay', 'publish': 'panel'} STOP = threading.Event() +EVENTS = False def digest(value): @@ -85,7 +87,8 @@ def validate(plan): if type(n['formal_group_id']) is not int or n['formal_group_id'] < 1: raise ValueError('An existing formal group is required') ipaddress.ip_address(n['node_address']) - ipaddress.ip_address(n['expected_exit']) + exit_address=ipaddress.ip_address(n['expected_exit']) + if getattr(exit_address,'scope_id',None):raise ValueError('Scoped/link-local exit addresses are not supported') for k in ('relay_host', 'tls_domain'): if not re.fullmatch(r'[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?\.[a-z]{2,}', n[k]): raise ValueError('Invalid DNS name: ' + k) @@ -143,16 +146,21 @@ class Ledger: CREATE TABLE IF NOT EXISTS batches(id TEXT PRIMARY KEY, digest TEXT NOT NULL); CREATE TABLE IF NOT EXISTS claims(namespace TEXT, kind TEXT, value TEXT, owner TEXT NOT NULL, PRIMARY KEY(namespace,kind,value)); + CREATE TABLE IF NOT EXISTS evidence(batch TEXT,node TEXT,phase TEXT, + dependency TEXT,expires REAL,request TEXT,PRIMARY KEY(batch,node,phase)); CREATE TABLE IF NOT EXISTS phases(batch TEXT, node TEXT, phase TEXT, status TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, updated REAL NOT NULL, PRIMARY KEY(batch,node,phase)); ''') os.chmod(self.path, 0o600) + @contextlib.contextmanager def db(self): c = sqlite3.connect(self.path, timeout=30) - c.execute('PRAGMA synchronous=FULL') - return c + try: + c.execute('PRAGMA synchronous=FULL') + with c: yield c + finally: c.close() def reserve(self, plan): namespace = plan['context']['panel_vps'] + ':' + plan['context']['app_root'] @@ -185,6 +193,15 @@ class Ledger: if old and old[0] != owner: raise ValueError('Another batch reserved the gateway SOCKS port') c.execute('INSERT OR IGNORE INTO claims VALUES(?,?,?,?)', (scope, 'socks_port', value, owner)) + formal_port=plan['context'].get('formal_socks_port') + if formal_port is not None: + if type(formal_port) is not int or not 1024<=formal_port<=65535 or any(n['socks_port']==formal_port for n in plan['nodes']): + raise ValueError('A distinct formal SOCKS port is required') + scope,value='gateway:'+plan['context']['gateway_vps'],str(formal_port) + owner=digest({'context':plan['context'],'formal_port':formal_port}) + old=c.execute('SELECT owner FROM claims WHERE namespace=? AND kind=? AND value=?',(scope,'socks_port',value)).fetchone() + if old and old[0]!=owner:raise ValueError('Formal SOCKS port already reserved') + c.execute('INSERT OR IGNORE INTO claims VALUES(?,?,?,?)',(scope,'socks_port',value,owner)) c.execute('INSERT OR IGNORE INTO batches VALUES(?,?)', (plan['batch_id'], digest(plan))) def state(self, batch, key, phase): @@ -193,11 +210,31 @@ class Ledger: (batch, key, phase)).fetchone() return row[0] if row else 'pending' + def reusable(self, batch, key, phase, dependency, request): + if not dependency or self.state(batch, key, phase) != 'done': + return False + with self.db() as c: + row = c.execute('SELECT dependency,expires,request FROM evidence WHERE batch=? AND node=? AND phase=?', (batch,key,phase)).fetchone() + return bool(row and row[0] == dependency and time.time() <= row[1] <= time.time()+900 and row[2] == request) + + def proof(self, batch, key, phase, dependency, expires, request): + with self.db() as c: + c.execute('INSERT OR REPLACE INTO evidence VALUES(?,?,?,?,?,?)',(batch,key,phase,dependency,expires,request)) + def mark(self, batch, key, phase, status, attempt=False): + previous=self.state(batch,key,phase) with self.db() as c: c.execute('''INSERT INTO phases VALUES(?,?,?,?,?,?) ON CONFLICT(batch,node,phase) DO UPDATE SET status=excluded.status,attempts=phases.attempts+?,updated=excluded.updated''', (batch, key, phase, status, int(attempt), time.time(), int(attempt))) + if previous!=status or attempt: + fd=os.open(self.root/'events.jsonl',os.O_CREAT|os.O_WRONLY|os.O_APPEND,0o600) + try: + os.fchmod(fd,0o600) + event=json.dumps({'batch':batch,'node':key,'phase':phase,'status':status,'time':time.time()})+'\n' + os.write(fd,event.encode()) + if EVENTS:sys.stdout.write(event);sys.stdout.flush() + finally:os.close(fd) class Coordinator: @@ -206,11 +243,35 @@ class Coordinator: self.ledger = Ledger(root) self.jobs = jobs self.timeout = timeout + self.dependencies = {} + self.runtime = None self.ledger.reserve(plan) self.dir = self.ledger.root / plan['batch_id'] self.dir.mkdir(mode=0o700, exist_ok=True) self.plan_file = self.dir / 'plan.json' atomic_json(self.plan_file, plan) + if plan.get('runtime'): + self.check_runtime() + directory = str(Path(__file__).resolve().parent) + if directory not in sys.path: sys.path.insert(0, directory) + from site_adapter import SiteAdapter + self.runtime = SiteAdapter(plan, self.dir, atomic_writer=atomic_json, resource_lock=lock) + + def check_runtime(self): + runtime = self.plan['runtime'] + if runtime.get('kind') != 'ssh-site-v1': raise ValueError('Unknown runtime') + if runtime.get('python') != str(Path(sys.executable).resolve()):raise ValueError('Use the frozen runtime interpreter') + import paramiko + if paramiko.__version__ != runtime.get('paramiko_version'):raise ValueError('Runtime dependency version changed') + base = Path(__file__).resolve().parent + required = [base/n for n in ('batch.py','site_adapter.py','ssh_pool.py','acceptance.py')] + required += [Path(self.plan['context'][k]) for k in ('ssh_config','exit_registry')] + files = runtime.get('files', {}) + if any(str(f) not in files for f in required): raise ValueError('Runtime source pin missing') + for f, sha in files.items(): + if not Path(f).is_absolute() or hashlib.sha256(Path(f).read_bytes()).hexdigest() != sha: + raise ValueError('Runtime code/config changed') + def call(self, n, phase, action): fingerprint = digest({'context': self.plan['context'], 'node': n}) @@ -218,6 +279,14 @@ class Coordinator: atomic_json(path, dict(n, request_sha256=fingerprint)) argv = self.plan['adapter'] + ['--plan', str(self.plan_file), '--node', str(path), '--phase', phase, '--action', action, '--state-dir', str(self.dir)] + if self.runtime: + self.check_runtime() + r = self.runtime.call(dict(n, request_sha256=fingerprint), phase, action) + if (r.get('key'),r.get('phase'),r.get('request_sha256')) != (n['key'],phase,fingerprint): + raise RuntimeError('Runtime identity mismatch') + if r.get('status') not in ('ready','absent','waiting','unknown','conflict'): + raise RuntimeError('Runtime status invalid') + return r with subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, start_new_session=True) as p: deadline = time.monotonic() + self.timeout @@ -258,8 +327,20 @@ class Coordinator: if resource in ('dns', 'relay'): namespace = 'dpl-external' with lock(self.ledger.root, namespace + ':' + resource): + dependency = self.dependencies.get(key, {}).get(phase) + request = digest({'plan': self.plan, 'node': n}) + # Publish/audit/preflight are always live; unknown or running phases cannot use a cache. + if phase in ('stage','dns','provision','relay','verify') and self.ledger.reusable(batch,key,phase,dependency,request): + if self.runtime: self.check_runtime() + return True r = self.call(n, phase, 'probe') + if phase == 'preflight' and r['status'] == 'ready': + deps = r.get('resume_dependencies', {}) + clean = {k:v for k,v in deps.items() if k in PHASES and isinstance(v,str) and re.fullmatch('[0-9a-f]{64}',v)} + self.dependencies[key] = {k:digest({p:clean.get(p) for p in PHASES[:PHASES.index(k)+1]}) for k in clean} if r['status'] == 'ready': + if dependency and type(r.get('valid_until')) in (int,float) and time.time() < r['valid_until'] <= time.time()+900: + self.ledger.proof(batch,key,phase,dependency,r['valid_until'],request) self.ledger.mark(batch, key, phase, 'done') return True if r['status'] == 'conflict': @@ -278,6 +359,8 @@ class Coordinator: return False r = self.call(n, phase, 'probe') self.ledger.mark(batch, key, phase, 'done' if r['status'] == 'ready' else r['status']) + if r['status'] == 'ready' and dependency and type(r.get('valid_until')) in (int,float) and time.time()banned && (int)$u->group_id===$group && (int)$u->u+(int)$u->d<(int)$u->transfer_enable && ($u->expired_at===null || (int)$u->expired_at>=time()); + } + public static function prepare(string $dir,int $group,bool $apply): array { + return locked($dir,function()use($dir,$group,$apply){ + if(!ServerGroup::find($group))throw new RuntimeException('Formal group missing'); + $path=$dir.'/acceptance-user.json'; + if(is_file($path) && readJson($path)['state']==='closed' && $apply){ + $old=readJson($path);$prior=isset($old['id'])?User::find($old['id']):null; + if($old['owned'] && $prior && (!$prior->banned || $prior->email!==$old['email'] || $prior->uuid!==$old['uuid'] || $prior->token!==$old['token'] || $prior->is_admin || (int)$prior->group_id!==$old['group']))throw new RuntimeException('Prior lease not revoked'); + writeJson($dir.'/acceptance-closed-'.($old['id']??'absent').'.json',$old);unlink($path); + } + if(is_file($path)){ + $j=readJson($path); + if($j['group']!==$group || $j['state']==='closed')throw new RuntimeException('Lease closed or parameters changed; never reactivate'); + $u=User::where('email',$j['email'])->first(); + if(!$u && $j['owned'] && $j['state']==='prepared' && $apply){ + $u=DB::transaction(fn()=>self::create($j),1); + } + if(!$u || $u->uuid!==$j['uuid'] || $u->token!==$j['token'] || !self::eligible($u,$group) || ($j['owned'] && $u->is_admin))throw new RuntimeException('Acceptance lease drifted'); + if($j['owned'])self::limits($u,$j); + $j['id']=$u->id;$j['state']='active';if($apply)writeJson($path,$j); + return ['state'=>'ready','id'=>$u->id,'owned'=>$j['owned']]; + } + $u=User::where('group_id',$group)->where('banned',0)->whereRaw('u+dwhere(fn($q)=>$q->whereNull('expired_at')->orWhere('expired_at','>=',time()))->where('email','not like','node-qa-%')->where('email','not like','batch-acceptance-%')->orderBy('id')->first(); + if(!$apply)return ['state'=>$u?'existing':'would_create']; + $j=['group'=>$group,'state'=>'prepared','owned'=>!$u,'email'=>$u?->email??'batch-acceptance-'.substr(hash('sha256',$dir),0,12).'-'.Str::lower(Str::random(8)).'@example.invalid','uuid'=>$u?->uuid??(string)Str::uuid(),'token'=>$u?->token??Str::random(32),'quota'=>104857600,'expires_at'=>time()+3600]; + // Persist identity before commit. A lost response resumes the same row, never another user. + if(!$u && User::where('email',$j['email'])->exists())throw new RuntimeException('Unowned acceptance email conflict'); + writeJson($path,$j); + if(!$u)$u=DB::transaction(fn()=>self::create($j),1); + $j['id']=$u->id;$j['state']='active';writeJson($path,$j); + return ['state'=>'ready','id'=>$u->id,'owned'=>$j['owned']]; + }); + } + private static function limits(User $u,array $j):void { + if((int)$u->transfer_enable!==$j['quota'] || (int)$u->expired_at!==$j['expires_at'] || (int)$u->device_limit!==1)throw new RuntimeException('Owned lease limits changed; inspect before mutation'); + } + private static function create(array $j): User { + $u=new User(['email'=>$j['email'],'uuid'=>$j['uuid'],'token'=>$j['token'],'password'=>password_hash(Str::random(40),PASSWORD_BCRYPT),'group_id'=>$j['group'],'transfer_enable'=>$j['quota'],'expired_at'=>$j['expires_at'],'banned'=>0,'is_admin'=>0,'remind_expire'=>0,'remind_traffic'=>0,'device_limit'=>1]);$u->saveQuietly();return $u; + } + public static function source(string $dir,int $group,bool $inspect=false): array { + $j=readJson($dir.'/acceptance-user.json');$u=User::findOrFail($j['id']); + if($j['group']!==$group || (!$inspect && $j['state']!=='active') || $u->uuid!==$j['uuid'] || $u->token!==$j['token'] || (!$inspect && !self::eligible($u,$group)) || ($inspect && !$j['owned'] && !self::eligible($u,$group)) || (int)$u->group_id!==$group)throw new RuntimeException('Acceptance user unavailable'); + if($j['owned'])self::limits($u,$j); + $nodes=[]; + foreach(ServerService::getAvailableServers($u) as $s){ + if(isset($s['ports']))throw new RuntimeException('Port-range nodes require an explicit verifier'); + $model=Server::findOrFail($s['id']); + if(!$inspect && !ServerService::getAvailableUsers($model)->contains('id',$u->id))throw new RuntimeException('User excluded by effective node filter'); + $nodes[]=['id'=>$s['id'],'type'=>$s['type'],'tag'=>$s['name'],'relay'=>$s['host'],'relay_port'=>(int)$s['port'],'sni'=>$s['protocol_settings']['tls']['server_name']??null,'uuid'=>$s['password']]; + } + if($inspect){foreach($nodes as &$n)unset($n['uuid']);unset($n);return ['nodes'=>$nodes];} + return ['user_id'=>$u->id,'owned'=>$j['owned'],'token'=>$u->token,'nodes'=>$nodes]; + } + public static function cleanup(string $dir,bool $apply): array { + return locked($dir,function()use($dir,$apply){ + $p=$dir.'/acceptance-user.json';if(!is_file($p))return ['state'=>'absent'];$j=readJson($p); + $u=isset($j['id'])?User::find($j['id']):User::where('email',$j['email'])->first(); + if($j['owned'] && $u){ + if($u->email!==$j['email'] || $u->uuid!==$j['uuid'] || $u->token!==$j['token'] || $u->is_admin || (int)$u->group_id!==$j['group'])throw new RuntimeException('Cleanup ownership mismatch'); + self::limits($u,$j); + if($apply){$u->banned=true;$u->saveQuietly();} + if($apply && !$u->fresh()->banned)throw new RuntimeException('Cleanup failed'); + } + if($apply){$j['state']='closed';$j['id']=$u?->id;writeJson($p,$j);} + return ['state'=>$apply?'closed':'preview','owned'=>$j['owned'],'id'=>$u?->id]; + }); + } +} +if(realpath($_SERVER['SCRIPT_FILENAME']??'')===__FILE__){ + $o=getopt('',['action:','dir:','group:','apply','help']); + if(isset($o['help'])){echo "--action=prepare|source|snapshot|cleanup --dir=PRIVATE_DIR --group=EXISTING_ID [--apply]\n";exit;} + try{ + $dir=$o['dir']??'';$group=(int)($o['group']??0); + if(!preg_match('~^/[A-Za-z0-9_./-]+$~D',$dir)||str_contains($dir,'..')||$group<1)throw new RuntimeException('Explicit private directory and group required'); + umask(0077);if(!is_dir($dir))mkdir($dir,0700,true); + if(is_link($dir)||(fileperms($dir)&0077))throw new RuntimeException('Private directory required'); + boot();$apply=isset($o['apply']); + $r=match($o['action']??''){'prepare'=>Acceptance::prepare($dir,$group,$apply),'source'=>Acceptance::source($dir,$group),'snapshot'=>Acceptance::source($dir,$group,true),'cleanup'=>Acceptance::cleanup($dir,$apply),default=>throw new RuntimeException('Unknown action')}; + echo json_encode($r,JSON_THROW_ON_ERROR).PHP_EOL; + }catch(\Throwable $e){fwrite(STDERR,json_encode(['state'=>'blocked','reason'=>'Inspect protected lease and effective authorization; no automatic overwrite']).PHP_EOL);exit(2);} +} diff --git a/scripts/node-onboarding/requirements.txt b/scripts/node-onboarding/requirements.txt new file mode 100644 index 0000000..e7f7b5d --- /dev/null +++ b/scripts/node-onboarding/requirements.txt @@ -0,0 +1 @@ +paramiko==5.0.0 diff --git a/scripts/node-onboarding/site_adapter.py b/scripts/node-onboarding/site_adapter.py new file mode 100644 index 0000000..b569b66 --- /dev/null +++ b/scripts/node-onboarding/site_adapter.py @@ -0,0 +1,431 @@ +"""Reusable AnyTLS site adapter; invoked in-process by batch.py, never a daemon.""" +import base64,hashlib,json,os,shlex,threading,time +from pathlib import Path +from types import SimpleNamespace +from ssh_pool import Pool,private_json +from acceptance import audit,digest,snapshot,service_change,provider_state + +class SiteAdapter: + def __init__(self,plan,directory,pool=None,atomic_writer=None,resource_lock=None): + if atomic_writer is None or resource_lock is None: + from batch import atomic_json,lock + atomic_writer,resource_lock=atomic_writer or atomic_json,resource_lock or lock + self.atomic_json,self.resource_lock=atomic_writer,resource_lock + self.plan,self.directory=plan,Path(directory) + self.ctx=plan['context'];self.lock=threading.RLock();self.leases={};self.formal_done={} + for field in ('ops_root','node_root','batch_backup_root','node_backup_root','qa_root','provider_snapshot','exit_registry','ssh_config','client_binary'): + value=self.ctx.get(field,'') + if not isinstance(value,str) or not value.startswith('/') or '..' in value.split('/') or any(c not in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_./-' for c in value): + raise ValueError('Explicit safe absolute site path required: '+field) + config=private_json(self.ctx['ssh_config']) + for candidate in plan['nodes']: + if config['hosts'].get(candidate['vps_id'],{}).get('host')!=candidate['node_address']:raise ValueError('SSH target differs from frozen candidate address') + for role in ('jump_vps','panel_vps','gateway_vps'): + if self.ctx[role] not in config['hosts']:raise ValueError('Missing registered SSH role') + self.pool=pool or Pool(config,self.ctx['jump_vps']) + if not __import__('re').fullmatch('[0-9a-f]{64}',self.ctx.get('client_sha256','')):raise ValueError('Reviewed official client SHA256 required') + self.exit_registry=private_json(self.ctx['exit_registry']) + for candidate in plan['nodes']: + if self.exit_registry.get(candidate['tls_domain'])!={'expected_exit':candidate['expected_exit'],'country':candidate['country']}:raise ValueError('Candidate exit registry missing or changed') + if any(n['vps_id'] in [self.ctx['jump_vps'],self.ctx['panel_vps'],self.ctx['gateway_vps']] for n in plan['nodes']):raise ValueError('Dedicated node host required') + for key in ('app_root',): + if not __import__('re').fullmatch('/[A-Za-z0-9_./-]+',self.ctx[key]) or '..' in self.ctx[key].split('/'):raise ValueError('Unsafe application root') + if type(self.ctx.get('formal_socks_port')) is not int or not 1024<=self.ctx['formal_socks_port']<=65535 or self.ctx['formal_socks_port'] in [n['socks_port'] for n in plan['nodes']]:raise ValueError('Reserve a distinct formal SOCKS port') + if not self.ctx.get('gateway_containers') or any(not __import__('re').fullmatch('[A-Za-z0-9_.-]+',v) for v in self.ctx['gateway_containers']):raise ValueError('Explicit gateway container inventory required') + + def close(self): + # On waiting, failure and success, close owned formal leases before closing SSH. + failed=False + for group in list(self.leases): + try:self.account_command(group,'cleanup',True) + except Exception:failed=True + self.pool.close() + if failed: + self.atomic_json(self.directory/'cleanup-error.json',{'groups':list(self.leases),'reason':'cleanup_outcome_unknown','needs_probe':True}) + raise RuntimeError('Acceptance cleanup incomplete; inspect protected leases before retry') + + def account_command(self,group,action,apply=False,client=None): + panel=client or self.pool.get(self.ctx['panel_vps']) + directory=self.ctx['batch_backup_root']+'/acceptance-group-'+str(group) + cmd='cd '+self.ctx['app_root']+' && php '+self.ctx['ops_root']+'/panel-acceptance.php --action='+action+' --dir='+directory+' --group='+str(group)+(' --apply' if apply else '') + _,out,err=panel.exec_command(cmd,timeout=90);data=out.read();err.read() + if out.channel.recv_exit_status():raise RuntimeError('acceptance_account_blocked') + try:return json.loads(data) + except ValueError:raise RuntimeError('acceptance_account_response_not_json') from None + + def prepare_account(self,n,panel,execute): + with self.lock: + group=n['formal_group_id'] + for spec in self.plan['nodes']: + registered=self.exit_registry.get(spec['tls_domain'],{}) + if registered != {'expected_exit':spec['expected_exit'],'country':spec['country']}: + raise ValueError('Exit registry must independently cover the candidate plan') + if group not in self.leases: + # Register cleanup obligation before any possibly successful remote create. + self.leases[group]=True + self.account_command(group,'prepare',True,panel) + source=self.account_command(group,'source',client=panel) + for spec in source['nodes']: + if spec['sni'] not in self.exit_registry:raise ValueError('Missing old-node exit expectation') + + def certificate_gate(self,n,node,read,execute): + owner=read(node,'/etc/dpl-node/batch-owner.json') + owned=bool(owner and json.loads(owner).get('request_sha256')==n['request_sha256']) + script=r"""from pathlib import Path +import re +for name in ['/etc/letsencrypt/cli.ini','/root/.config/letsencrypt/cli.ini']: + p=Path(name) + if p.exists(): + for line in p.read_text().splitlines(): + if re.match(r'^\s*(pre[-_]hook|post[-_]hook|deploy[-_]hook)\s*=\s*\S',line):raise SystemExit(2) +domain=DOMAIN +paths=[Path('/etc/letsencrypt')/kind/domain for kind in ['live','archive']]+[Path('/etc/letsencrypt/renewal')/(domain+'.conf')] +if not OWNED and any(p.exists() or p.is_symlink() for p in paths):raise SystemExit(3) +p=paths[-1] +if p.exists(): + for line in p.read_text().splitlines(): + if re.match(r'^\s*(pre_hook|post_hook)\s*=\s*\S',line):raise SystemExit(4) + if re.match(r'^\s*deploy_hook\s*=',line) and line.split('=',1)[1].strip()!=HOOK:raise SystemExit(5) +""" + prefix='DOMAIN='+repr(n['tls_domain'])+';OWNED='+repr(owned)+';HOOK='+repr('/bin/sh '+self.ctx['node_root']+'/renew-node.sh')+'\n' + execute(node,"python3 - <<'DPLCERTGATE'\n"+prefix+script+"\nDPLCERTGATE") + execute(node,'command -v git >/dev/null && command -v curl >/dev/null && command -v certbot >/dev/null && command -v ss >/dev/null && command -v sha256sum >/dev/null && command -v openssl >/dev/null') + + def guard_service(self,node,backup,read,write,execute): + spec=self.ctx['preserved_service'] + for field in ('unit','binary','config','port'): + if field not in spec:raise ValueError('Explicit preserved service specification required') + if not all(isinstance(spec[k],str) and __import__('re').fullmatch('[A-Za-z0-9_./@-]+',spec[k]) for k in ('unit','binary','config')) or type(spec['port']) is not int: + raise ValueError('Unsafe preserved service specification') + script="""import hashlib,json,pathlib,subprocess +s=SPEC +v=dict(x.split('=',1) for x in subprocess.check_output(['systemctl','show',s['unit'],'-p','MainPID','-p','ActiveEnterTimestamp','-p','NRestarts','-p','ActiveState','-p','FragmentPath'],text=True).splitlines()) +def h(p):return hashlib.sha256(pathlib.Path(p).read_bytes()).hexdigest() +listeners=subprocess.check_output(['ss','-lntH','( sport = :'+str(s['port'])+' )'],text=True) +print(json.dumps({'unit_sha256':h(v['FragmentPath']),'binary_sha256':h(s['binary']),'config_sha256':h(s['config']),'listeners':sorted(x.split()[3] for x in listeners.splitlines()),'pid':v['MainPID'],'started':v['ActiveEnterTimestamp'],'restarts':v['NRestarts'],'active':v['ActiveState']=='active'})) +""".replace('SPEC',repr(spec)) + current=json.loads(execute(node,"python3 - <<'DPLSERVICE'\n"+script+"\nDPLSERVICE")) + if not current['active'] or not current['listeners']:raise RuntimeError('preserved_service_unavailable') + execute(node,'install -d -m 700 '+backup) + p=backup+'/preserved-service-original.json';raw=read(node,p) + if raw is None:write(node,p,json.dumps(current));return current + before=json.loads(raw);status=service_change(before,current) + if status=='investigate_restart': + write(node,backup+'/preserved-service-observed.json',json.dumps(current)) + review=read(node,backup+'/preserved-service-review.json') + review=json.loads(review) if review else {} + if review.get('before_sha256')!=digest(before) or review.get('after_sha256')!=digest(current) or not review.get('reason') or not review.get('evidence_sha256'): + raise RuntimeError('investigate_preserved_service_restart') + elif status!='ready':raise RuntimeError('preserved_service_'+status) + return current + + def dependencies(self,n,panel,gateway,node,stage_probe,read,execute,gateway_py): + """Current dependency reads; a missing observation cannot authorize a skip.""" + state=stage_probe(panel);deps={'stage':digest(state)} + ui=provider_state(private_json(self.ctx['provider_snapshot']),n) + try: + dns=json.loads(gateway_py(gateway,'import socket,json\nprint(json.dumps(sorted(set(x[4][0] for x in socket.getaddrinfo('+repr(n['tls_domain'])+',443,socket.AF_INET)))))')) + if dns==[n['node_address']]:deps['dns']=digest(dns) + except Exception:pass + try: + live=execute(node,'systemctl is-active dpl-node; git -C '+self.ctx['node_root']+' rev-parse HEAD; sha256sum /etc/dpl-node/config.yml /etc/dpl-node/secret.env /etc/systemd/system/dpl-node.service /etc/letsencrypt/live/'+n['tls_domain']+'/fullchain.pem') + deps['provision']=digest(live) + # End-to-end strict TLS at preflight is necessary to reuse a relay observation. + tls=gateway_py(gateway,'import ssl,socket\nc=ssl.create_default_context()\nwith socket.create_connection(('+repr(n['relay_host'])+','+str(n['relay_port'])+'),10) as s:\n with c.wrap_socket(s,server_hostname='+repr(n['tls_domain'])+') as t:print(t.version())') + rows=ui['relays'] + deps['relay']=digest([rows,tls]) + report=read(panel,self.ctx['node_backup_root']+'/'+n['key']+'/qa-report.json') + if report:deps['verify']=digest([state,live,hashlib.sha256(report).hexdigest()]) + except Exception:pass + return deps + + def formal(self,n,panel,gateway,php,execute,read,write): + group=n['formal_group_id'];ctx=self.ctx + with self.lock, self.resource_lock(self.directory.parent, ctx['gateway_vps']+':formal:'+str(ctx['formal_socks_port'])): + source=self.account_command(group,'snapshot' if group in self.formal_done else 'source',client=panel) + def public(source): + values=[] + for item in source['nodes']: + if item['sni'] not in self.exit_registry:raise ValueError('Unregistered current node exit') + values.append(dict(item,**self.exit_registry[item['sni']])) + return snapshot(values) + before=public(source) + # Do not run final verification while any candidate in this group is still isolated. + cohort=[x for x in self.plan['nodes'] if x['formal_group_id']==group] + wanted={(x['relay_host'],x['relay_port'],x['tls_domain']) for x in cohort} + if not wanted.issubset({(x['relay'],x['relay_port'],x['sni']) for x in before}):raise RuntimeError('formal_group_not_ready') + if group in self.formal_done: + prior=self.formal_done[group] + return audit(prior['before'],before,prior['report'],prior['qas'],prior['binding'],digest(self.plan),prior['expected_qas']) + root=ctx['qa_root']+'/formal-group-'+str(group) + execute(gateway,'install -d -m 700 '+root) + expectations=[] + for node in before: + original=next(x for x in source['nodes'] if x['id']==node['id']) + expectations.append({'tag':node['tag'],'host':node['relay'],'port':node['relay_port'],'sni':node['sni'],'uuid':original['uuid'],'expected_exit':node['expected_exit'],'expected_country':node['country']}) + # Tokens/UUIDs never leave memory except the gateway's private task files. + write(gateway,root+'/source-private.json',json.dumps({'token':source['token'],'expectations':expectations,'owned':source['owned']})) + script="""import gzip,json,urllib.request,subprocess,os,time +from pathlib import Path +os.umask(0o077);root=Path(ROOT);source=json.loads((root/'source-private.json').read_text()) +r=urllib.request.Request(SUB+'/api/v1/client/subscribe?token='+source['token'],headers={'User-Agent':'sing-box/1.14.3'}) +with urllib.request.urlopen(r,timeout=30) as response: + data=response.read();headers={k.lower():v for k,v in response.headers.items()} + if headers.get('content-encoding','').lower()=='gzip':data=gzip.decompress(data) +assert headers.get('cf-cache-status','').upper() not in ['HIT','STALE','UPDATING'] and not headers.get('age') and not headers.get('cf-mitigated') and 'no-store' in headers.get('cache-control','').lower() +(root/'subscription.json').write_bytes(data);(root/'expectations.json').write_text(json.dumps(source['expectations'])) +deadline=time.monotonic()+900 +for attempt in range(3 if source['owned'] else 1): + with (root/'verify.stdout').open('wb') as out,(root/'verify.stderr').open('wb') as err: + result=subprocess.run(['python3',OPS+'/verify-node.py','--subscription',str(root/'subscription.json'),'--expectations',str(root/'expectations.json'),'--binary',BINARY,'--port',str(PORT),'--report',str(root/'report.json'),'--run'],timeout=max(1,deadline-time.monotonic()),stdout=out,stderr=err) + if result.returncode==0:break + try:reason=json.loads((root/'verify.stderr').read_text()).get('reason') + except Exception:reason=None + if not source['owned'] or attempt==2 or reason!='Verification failed; inspect protected diagnostics' or deadline-time.monotonic()<15:raise RuntimeError('Formal verification failed') + # A just-created user's UUID may need one node pull interval to propagate. + # Retry only diagnostic reads, keep the same lease; never replay production writes. + time.sleep(15) +""" + prefix='ROOT='+repr(root)+';SUB='+repr(ctx['subscription_url'])+';OPS='+repr(ctx['ops_root'])+';BINARY='+repr(ctx['client_binary'])+';PORT='+repr(ctx['formal_socks_port'])+'\n' + try: + execute(gateway,"python3 - <<'DPLFORMAL'\n"+prefix+script+"\nDPLFORMAL",1000) + after=public(self.account_command(group,'source',client=panel)) + report=read(gateway,root+'/report.json') + finally: + # Never leave an owned formal test user active, including failed client runs. + self.account_command(group,'cleanup',True,panel) + identities={};tokens={} + for candidate in cohort: + b=ctx['node_backup_root']+'/'+candidate['key'] + result=php(panel,'$m=json_decode(file_get_contents('+self.php_value(b+'/manifest.json')+'),true);$s=App\\Models\\Server::findOrFail($m["node_id"]);$u=App\\Models\\User::findOrFail($m["test_user_id"]);echo json_encode(["id"=>$u->id,"token"=>$u->token,"banned"=>(bool)$u->banned,"excluded"=>!App\\Services\\ServerService::getAvailableUsers($s)->contains("id",$u->id)]);') + identities[candidate['key']]=result['id'];tokens[candidate['key']]=result + # Include every owned formal lease from earlier resumptions, not just the current user. + lease_dir=ctx['batch_backup_root']+'/acceptance-group-'+str(group) + leases=php(panel,'$out=[];foreach(glob('+self.php_value(lease_dir+'/acceptance-*.json')+') as $f){$j=json_decode(file_get_contents($f),true);if(!($j["owned"]??false)||empty($j["id"]))continue;$u=App\\Models\\User::findOrFail($j["id"]);$excluded=true;foreach(App\\Models\\Server::all() as $s)if(App\\Services\\ServerService::getAvailableUsers($s)->contains("id",$u->id))$excluded=false;$out[(string)$u->id]=["id"=>$u->id,"token"=>$u->token,"banned"=>(bool)$u->banned,"excluded"=>$excluded];}echo json_encode($out);') + for ident,value in (leases.items() if isinstance(leases,dict) else []): + key='formal-'+str(ident);identities[key]=value['id'];tokens[key]=value + users={} + for key,value in tokens.items(): + # Only the gateway accesses the subscription source; local HTTP never bypasses its registered route. + script='import urllib.request,urllib.error\nr=urllib.request.Request('+repr(ctx['subscription_url']+'/api/v1/client/subscribe?token='+value['token'])+',headers={"User-Agent":"sing-box/1.14.3"})\ntry:\n response=urllib.request.urlopen(r,timeout=30);print(response.status);response.close()\nexcept urllib.error.HTTPError as e:print(e.code)' + # Keep tokens out of argv: exec_command carries a private stdin script, not an OS argv. + si,o,e=gateway.exec_command('python3 -',timeout=60);si.write(script);si.channel.shutdown_write();out=o.read();e.read() + if o.channel.recv_exit_status():raise RuntimeError('qa_rejection_probe_failed') + users[key]={'id':value['id'],'banned':value['banned'],'excluded':value['excluded'],'http_status':int(out.strip())} + qas=json.dumps({'users':users,'captured_at':int(time.time()),'plan_sha256':digest(self.plan)},sort_keys=True).encode() + binding={'plan_sha256':digest(self.plan),'snapshot_sha256':digest(before),'report_sha256':hashlib.sha256(report).hexdigest(),'qa_rejections_sha256':hashlib.sha256(qas).hexdigest(),'verified_at':int(time.time())} + audit(before,after,report,qas,binding,digest(self.plan),identities) + write(gateway,root+'/qa-rejections.json',qas);write(gateway,root+'/binding.json',json.dumps(binding));write(gateway,root+'/snapshot.json',json.dumps(before)) + self.formal_done[group]={'before':before,'report':report,'qas':qas,'binding':binding,'expected_qas':identities} + # Persist only nonsecret, byte-bound results for coordinator/operator review. + self.atomic_json(self.directory/('formal-group-'+str(group)+'.json'),{'snapshot':before,'report':json.loads(report),'qas':json.loads(qas),'binding':binding}) + + @staticmethod + def php_value(value): + return "json_decode(base64_decode('"+base64.b64encode(json.dumps(value).encode()).decode()+"'),true)" + + def call(self,n,phase,action): + plan=self.plan;ctx=self.ctx;key=n['key'];sha=n['request_sha256'] + a=SimpleNamespace(phase=phase,action=action,state_dir=str(self.directory)) + B=ctx['node_backup_root']+'/'+key;Q=ctx['qa_root']+'/'+key + OPS=ctx['ops_root'];NP=ctx['node_root'];APP=ctx['app_root'] + def execute(c,cmd,timeout=120): + si,o,e=c.exec_command(cmd,timeout=timeout);out=o.read();err=e.read();code=o.channel.recv_exit_status() + if code: raise RuntimeError('remote_exit_'+str(code)) + return out.decode() + def php(c,body): + si,o,e=c.exec_command('cd '+shlex.quote(APP)+' && php',timeout=90) + si.write("make(Illuminate\\Contracts\\Console\\Kernel::class)->bootstrap();"+body) + si.channel.shutdown_write();out=o.read();err=e.read() + if o.channel.recv_exit_status():raise RuntimeError('panel_php_failed') + try:return json.loads(out) + except ValueError:raise RuntimeError('panel_response_not_json') from None + def payload(value):return "json_decode(base64_decode('"+base64.b64encode(json.dumps(value,ensure_ascii=False).encode()).decode()+"'),true)" + def read(c,path): + with c.open_sftp() as s: + try: + with s.open(path,'rb') as f:return f.read() + except FileNotFoundError:return None + except OSError as e: + if e.errno==2:return None + raise + def write(c,path,data): + if isinstance(data,str):data=data.encode() + with c.open_sftp() as s: + tmp=path+'.tmp-'+str(os.getpid())+'-'+str(threading.get_ident()) + with s.open(tmp,'w') as f:s.chmod(tmp,0o600);f.write(data) + s.posix_rename(tmp,path) + def parameters():return dict(sorted({'key':key,'name':n['name'],'relay-host':n['relay_host'],'relay-port':n['relay_port'],'listen-port':n['listen_port'],'tls-domain':n['tls_domain'],'panel-url':ctx['panel_url']}.items())) + def canonical(value):return hashlib.sha256(json.dumps(value,sort_keys=True,ensure_ascii=False,separators=(',',':')).encode()).hexdigest() + def containers(gateway): + data=json.loads(execute(gateway,'docker inspect '+ ' '.join(ctx['gateway_containers']))) + return {v['Name']:{'id':v['Id'],'started':v['State']['StartedAt'],'restarts':v['RestartCount']} for v in data} + def receipt(status,retry=False):return {'key':key,'phase':a.phase,'request_sha256':sha,'status':status,'valid_until':time.time()+300,**({'retry_safe':True} if retry else {})} + def stage_probe(panel): + p=parameters() + code='$b='+payload(B)+';$p='+payload(p)+r''';$f=$b.'/manifest.json';if(!is_file($f)){$conf=App\Models\ServerMachine::where('name','dpl-'.$p['key'])->exists()||App\Models\Server::where('name',$p['name'])->exists()||App\Models\Server::where('host',$p['relay-host'])->where('port',(string)$p['relay-port'])->exists();echo json_encode(['status'=>$conf?'conflict':'absent','retry_safe'=>!$conf]);return;}$m=json_decode(file_get_contents($f),true);$s=App\Models\Server::find($m['node_id']);$mch=App\Models\ServerMachine::find($m['machine_id']);$ok=($m['parameters']??null)===$p && $s && $mch && $s->machine_id===$mch->id && $mch->name==='dpl-'.$p['key'] && $s->type==='anytls' && ($s->protocol_settings['tls']['allow_insecure']??null)===false && $s->name===$p['name'] && $s->host===$p['relay-host'] && (int)$s->port===$p['relay-port'] && (int)$s->server_port===$p['listen-port'] && ($s->protocol_settings['tls']['server_name']??null)===$p['tls-domain'];$u=App\Models\User::find($m['test_user_id']);$g=App\Models\ServerGroup::find($m['group_id']);$ok=$ok && $u && $g && !$u->is_admin && (int)$u->group_id===$m['group_id'] && str_starts_with($u->email,'node-qa-'.$p['key'].'-') && $g->name==='dpl 接入验收 '.$p['key'];if(!is_file($b.'/promotion-journal.json'))$ok=$ok && $s->group_ids===[(string)$m['group_id']] && !$u->banned;echo json_encode(['status'=>$ok?'ready':'conflict','manifest'=>$m]);''' + return php(panel,code) + def state(panel): + return php(panel,'$b='+payload(B)+r''';$m=json_decode(file_get_contents($b.'/manifest.json'),true);$s=App\Models\Server::findOrFail($m['node_id']);$u=App\Models\User::findOrFail($m['test_user_id']);echo json_encode(['manifest'=>$m,'node_online'=>(bool)$s->is_online,'groups'=>$s->group_ids,'qa_banned'=>(bool)$u->banned,'u'=>(int)$u->u,'d'=>(int)$u->d]);''') + def baseline(panel,gateway): + directory=ctx['batch_backup_root'];f=directory+'/baseline.json' + version=json.loads(execute(panel,"python3 - <<'DPLVERSION'\nimport subprocess,hashlib,json\np="+repr(APP)+"\nh=subprocess.check_output(['git','-C',p,'rev-parse','HEAD']).decode().strip()\nd=subprocess.check_output(['git','-C',p,'diff','HEAD'])+subprocess.check_output(['git','-C',p,'status','--porcelain'])\nprint(json.dumps({'head':h,'worktree_sha256':hashlib.sha256(d).hexdigest()}))\nDPLVERSION")) + previous=read(panel,directory+'/application-version.json') + if previous is None:write(panel,directory+'/application-version.json',json.dumps(version)) + elif json.loads(previous)!=version:raise RuntimeError('application_version_changed') + # Read/write a protected snapshot only once, before any new stage. + return php(panel,'$p='+payload(f)+r''';if(is_file($p)){echo json_encode(['exists'=>true]);return;} $v=['nodes'=>App\Models\Server::orderBy('id')->get()->toArray(),'plans'=>App\Models\Plan::orderBy('id')->get()->toArray(),'users'=>App\Models\User::orderBy('id')->get(['id','group_id','plan_id'])->toArray(),'subscribe_url'=>admin_setting('subscribe_url')];file_put_contents($p,json_encode($v,JSON_THROW_ON_ERROR));chmod($p,0600);echo json_encode(['exists'=>true]);''') + def gateway_py(gateway,script,timeout=120): + return execute(gateway,"python3 - <<'DPLPY'\n"+script+"\nDPLPY",timeout) + def assert_old(node): + self.guard_service(node,B,read,write,execute) + def transfer(panel,gateway,node): + execute(node,'install -d -m 700 /etc/dpl-node') + execute(gateway,'install -d -m 700 '+Q) + for source,dest,target in [('node-config.yml',node,'/etc/dpl-node/config.yml'),('node-secret.env',node,'/etc/dpl-node/secret.env'),('qa-private.json',gateway,Q+'/qa.json')]: + data=read(panel,B+'/'+source);prior=read(dest,target) + if prior is not None and prior!=data:raise RuntimeError('private_config_conflict') + if prior is None:write(dest,target,data) + def fetch_qa(gateway): + script='''import json,urllib.request + +from pathlib import Path + +q=json.loads(Path(Q+'/qa.json').read_text()) + +req=urllib.request.Request(SUB+'/api/v1/client/subscribe?token='+q['subscribe_token'],headers={'User-Agent':'sing-box/1.14.3'}) + +with urllib.request.urlopen(req,timeout=30) as r: + + data=r.read(); code=r.status; headers={k.lower():v for k,v in r.headers.items()} + +import gzip + +if headers.get('content-encoding','').lower()=='gzip':data=gzip.decompress(data) + +v=json.loads(data);Path(Q+'/subscription.json').write_bytes(data);Path(Q+'/subscription.json').chmod(0o600) + +print(json.dumps({'status':code,'cache':headers.get('cf-cache-status'),'control':headers.get('cache-control'),'age':headers.get('age')})) + +'''.replace('\n+','\n') + return gateway_py(gateway,'Q='+repr(Q)+';SUB='+repr(ctx['subscription_url'])+'\n'+script) + def phase_impl(panel,gateway,node): + if a.phase=='preflight': + with self.lock, self.resource_lock(self.directory.parent,ctx['panel_vps']+':panel'): + execute(panel,'install -d -m 700 '+ctx['batch_backup_root']) + execute(panel,'test "$(git -C '+OPS+' rev-parse HEAD)" = '+ctx['provision_commit']+' && test -z "$(git -C '+OPS+' status --porcelain)"') + baseline(panel,gateway) + base_path=ctx['batch_backup_root']+'/gateway-baseline.json' + current_containers=containers(gateway) + prior_gateway=read(panel,base_path) + if prior_gateway is None:write(panel,base_path,json.dumps(current_containers,sort_keys=True)) + elif json.loads(prior_gateway)!=current_containers:return receipt('conflict') + ui=Path(ctx['provider_snapshot']) + if not ui.exists():return receipt('waiting') + snap=private_json(ui) + if type(snap.get('captured_at')) not in (int,float) or not 0<=time.time()-snap['captured_at']<=900:return receipt('waiting') + try:provider_state(snap,n) + except ValueError:return receipt('conflict') + gateway_py(gateway,'import socket\ns=socket.socket();s.bind(("127.0.0.1",'+str(n['socks_port'])+'));s.close()') + s=stage_probe(panel) + if s['status']=='conflict':return receipt('conflict') + assert_old(node) + self.certificate_gate(n,node,read,execute) + execute(node,'test "$(git -C '+NP+' rev-parse HEAD)" = '+ctx['provision_commit']+' && test -z "$(git -C '+NP+' status --porcelain)" && command -v git >/dev/null') + execute(gateway,"echo '"+ctx['client_sha256']+" "+ctx['client_binary']+"' | sha256sum -c --status && test -x "+shlex.quote(ctx['client_binary'])+' && test "$(git -C '+OPS+' rev-parse HEAD)" = '+ctx['provision_commit']+' && test -z "$(git -C '+OPS+' status --porcelain)"') + if s['status']=='absent':execute(node,"test ! -e /etc/dpl-node/config.yml && test ! -e /etc/systemd/system/dpl-node.service && test ! -e /etc/dpl-node/batch-owner.json && ! ss -lntH '( sport = :"+str(n['listen_port'])+" )' | grep -q .") + if not php(panel,'echo json_encode(App\\Models\\ServerGroup::where("id",'+str(n['formal_group_id'])+')->exists());'):return receipt('conflict') + self.prepare_account(n,panel,execute) + r=receipt('ready');r['resume_dependencies']=self.dependencies(n,panel,gateway,node,stage_probe,read,execute,gateway_py);return r + if a.phase=='stage': + if a.action=='probe': + r=stage_probe(panel);return receipt(r['status'],r.get('retry_safe',False)) + command='cd '+APP+' && php '+OPS+'/stage.php '+ ' '.join('--'+k+'='+shlex.quote(str(v)) for k,v in {**parameters(),'backup':B}.items()) + execute(panel,command);return receipt('ready') + if a.phase=='dns': + try: + result=gateway_py(gateway,'import socket,json\nprint(json.dumps(sorted(set(x[4][0] for x in socket.getaddrinfo('+repr(n['tls_domain'])+',443,socket.AF_INET)))))') + return receipt('ready' if json.loads(result)==[n['node_address']] else 'conflict') + except Exception:return receipt('waiting') + if a.phase=='provision': + owner='/etc/dpl-node/batch-owner.json';prior=read(node,owner) + if prior is not None and json.loads(prior).get('request_sha256')!=sha:return receipt('conflict') + if prior is None and read(node,'/etc/systemd/system/dpl-node.service') is not None:return receipt('conflict') + done=read(node,B+'/provision-ok.json') + if a.action=='probe': + if done and json.loads(done).get('request_sha256')==sha: + for file,target in [('node-config.yml','/etc/dpl-node/config.yml'),('node-secret.env','/etc/dpl-node/secret.env')]: + if read(panel,B+'/'+file)!=read(node,target):return receipt('conflict') + execute(node,"echo '55bf71fa9d9f2048d3255ae7c0af929a41897ca7743f6ead34132a6ca4c79043 /opt/dpl-node/xboard-node' | sha256sum -c >/dev/null") + assert_old(node);execute(node,'systemctl is-active --quiet dpl-node && openssl x509 -checkend 86400 -noout -in /etc/letsencrypt/live/'+n['tls_domain']+'/fullchain.pem') + return receipt('ready') + return receipt('absent',True) + if prior is None: + execute(node,'install -d -m 700 /etc/dpl-node');write(node,owner,json.dumps({'request_sha256':sha,'key':key})) + transfer(panel,gateway,node) + if read(node,'/etc/systemd/system/dpl-node.service') is None: + execute(node,'bash '+NP+'/install-node.sh --listen-port='+str(n['listen_port'])+' --tls-domain='+n['tls_domain']+' > '+B+'/install-node.log 2>&1',300) + execute(node,"echo '55bf71fa9d9f2048d3255ae7c0af929a41897ca7743f6ead34132a6ca4c79043 /opt/dpl-node/xboard-node' | sha256sum -c >/dev/null") + domain=n['tls_domain'] + if read(node,'/etc/letsencrypt/live/'+domain+'/fullchain.pem') is None: + execute(node,"certbot certonly --standalone --non-interactive --agree-tos --no-directory-hooks --cert-name "+domain+' -d '+domain+" --deploy-hook '/bin/sh "+NP+"/renew-node.sh' > "+B+'/certbot.log 2>&1',300) + execute(node,'systemctl enable --now dpl-node > '+B+'/enable.log 2>&1 && systemctl is-active --quiet dpl-node',60) + execute(node,'certbot renew --cert-name '+domain+' --dry-run --no-random-sleep-on-renew --run-deploy-hooks --no-directory-hooks > '+B+'/renew-dry-run.log 2>&1',600) + assert_old(node) + write(node,B+'/provision-ok.json',json.dumps({'request_sha256':sha,'verified_at':int(time.time())})) + return receipt('ready') + if a.phase=='relay': + try: + result=gateway_py(gateway,'import ssl,socket,json\nc=ssl.create_default_context()\nwith socket.create_connection(('+repr(n['relay_host'])+','+str(n['relay_port'])+'),10) as s:\n with c.wrap_socket(s,server_hostname='+repr(n['tls_domain'])+') as t: print(json.dumps({"tls":t.version()}))') + return receipt('ready' if json.loads(result).get('tls') else 'waiting') + except Exception:return receipt('waiting') + if a.phase=='verify': + s=state(panel) + if s['qa_banned'] and s['groups']==[str(n['formal_group_id'])]: + return receipt('ready' if read(panel,B+'/qa-report.json') else 'unknown') + existing=read(panel,B+'/qa-report.json') + if existing: + report=json.loads(existing) + if report.get('state')=='live_verified' and report.get('qa_user_id')==s['manifest']['test_user_id'] and 0<=time.time()-report.get('verified_at',0)<=900 and s['node_online'] and s['u']+s['d']>0: + r=receipt('ready');r['valid_until']=min(r['valid_until'],report['verified_at']+900);return r + transfer(panel,gateway,node);fetch_qa(gateway) + execute(gateway,'python3 '+OPS+'/verify-node.py --qa '+Q+'/qa.json --subscription '+Q+'/subscription.json --binary '+ctx['client_binary']+' --port '+str(n['socks_port'])+' --expected-exit '+shlex.quote(n['expected_exit'])+' --expected-country '+n['country']+' --report '+Q+'/qa-report.json --run > '+Q+'/verify.stdout 2> '+Q+'/verify.stderr',300) + data=read(gateway,Q+'/qa-report.json');report=json.loads(data) + write(panel,B+'/qa-report.json',data) + # Reports contain endpoints and measured results, no passwords or tokens. + evidence={'key':key,'request_sha256':sha,'report':report,'panel':state(panel)} + (Path(a.state_dir)/(key+'-evidence.json')).write_text(json.dumps(evidence,ensure_ascii=False,indent=2)) + for _ in range(24): + s=state(panel) + if s['node_online'] and s['u']+s['d']>0: + r=receipt('ready');r['valid_until']=min(r['valid_until'],report['verified_at']+900);return r + time.sleep(5) + return receipt('waiting') + if a.phase=='publish': + s=state(panel) + report_data=read(panel,B+'/qa-report.json') + review_path=Path(a.state_dir)/(key+'-reviewed.json') + if not report_data or not review_path.exists():return receipt('waiting') + report=json.loads(report_data);review=json.loads(review_path.read_text()) + valid=review.get('request_sha256')==sha and review.get('report_sha256')==hashlib.sha256(report_data).hexdigest() and review.get('qa_user_id')==s['manifest']['test_user_id'] and review.get('verified_at')==report.get('verified_at') + if not valid:return receipt('waiting') + if s['qa_banned'] and s['groups']==[str(n['formal_group_id'])]: + journal=read(panel,B+'/promotion-journal.json') + if not journal:return receipt('conflict') + j=json.loads(journal) + return receipt('ready' if j.get('manifest')==s['manifest'] and j.get('request',{}).get('mode')=='append' and j.get('request',{}).get('group')==n['formal_group_id'] and j.get('state')=='applied' else 'conflict') + if not s['node_online'] or s['u']+s['d']<=0:return receipt('waiting') + if a.action=='probe':return receipt('absent',True) + cmd='cd '+APP+' && php '+OPS+'/promote.php --backup='+B+' --mode=append --group='+str(n['formal_group_id'])+' --verification='+B+'/qa-report.json' + execute(panel,cmd);execute(panel,cmd+' --apply');return receipt('ready') + if a.phase=='audit': + s=state(panel);assert_old(node) + if not s['qa_banned'] or s['groups']!=[str(n['formal_group_id'])] or not s['node_online']:return receipt('conflict') + ok=php(panel,'$b='+payload(B)+r''';$m=json_decode(file_get_contents($b.'/manifest.json'),true);$n=App\Models\Server::findOrFail($m['node_id']);$base=json_decode(file_get_contents('''+payload(ctx['batch_backup_root']+'/baseline.json')+r'''),true);$ok=$base['plans']===App\Models\Plan::orderBy('id')->get()->toArray() && $base['users']===App\Models\User::whereIn('id',array_column($base['users'],'id'))->orderBy('id')->get(['id','group_id','plan_id'])->toArray() && $base['subscribe_url']===admin_setting('subscribe_url');foreach($base['nodes'] as $old){$cur=App\Models\Server::find($old['id'])->toArray();foreach(['name','host','port','server_port','protocol_settings','group_ids','machine_id','enabled','show','cert_config'] as $k)if(($cur[$k]??null)!==($old[$k]??null))$ok=false;}$ids=App\Services\ServerService::getAvailableUsers($n)->pluck('id')->all();$ok=$ok&&!in_array($m['test_user_id'],$ids);echo json_encode($ok);''') + if not ok:return receipt('conflict') + self.formal(n,panel,gateway,php,execute,read,write) + original_gateway=read(panel,ctx['batch_backup_root']+'/gateway-baseline.json') + if not original_gateway or containers(gateway)!=json.loads(original_gateway):return receipt('conflict') + return receipt('ready') + return receipt('unknown') + panel=Lazy(self.pool,ctx['panel_vps']);gateway=Lazy(self.pool,ctx['gateway_vps']);node=Lazy(self.pool,n['vps_id']) + try:return phase_impl(panel,gateway,node) + except Exception as exc: + # Never print underlying library/server errors or credentials. + safe = str(exc) if type(exc) is RuntimeError and __import__('re').fullmatch('[a-z_0-9]+',str(exc)) else 'inspect_protected_state' + self.atomic_json(self.directory/(key+'-'+phase+'-error.json'),{'reason':safe}) + return receipt('waiting' if safe in ('investigate_preserved_service_restart','formal_group_not_ready') else 'unknown') + +class Lazy: + def __init__(self,pool,rid):self.pool,self.rid=pool,rid + def __getattr__(self,name):return getattr(self.pool.get(self.rid),name) diff --git a/scripts/node-onboarding/ssh_pool.py b/scripts/node-onboarding/ssh_pool.py new file mode 100644 index 0000000..35b602a --- /dev/null +++ b/scripts/node-onboarding/ssh_pool.py @@ -0,0 +1,99 @@ +"""Thread-local SSH sessions, mandatory explicit jump, strict known-host keys.""" +import json +import hashlib +import base64 +import re +import threading +from pathlib import Path + + +def private_text(path): + p = Path(path) + if not p.is_absolute() or p.is_symlink() or not p.is_file() or p.stat().st_mode & 0o077: + raise ValueError('Private regular absolute credential/config reference required') + return p.read_text() + + +def private_json(path): + return json.loads(private_text(path)) + + +def credentials(spec): + if not spec.get('credential_ref'):return {} + if spec.get('credential_format','json')=='json':return private_json(spec['credential_ref']) + if spec['credential_format']=='codocs-password': + matches=re.findall(r'(?m)^[ \t]*(?:[-*][ \t]+)?密码[：:][ \t]*([^\r\n]+)$',private_text(spec['credential_ref'])) + if len(matches)!=1:raise ValueError('Unambiguous existing CoDocs password field required') + return {'password':matches[0].strip()} + raise ValueError('Unsupported credential reference format') + + +class Pool: + def __init__(self, config, jump_id, client_factory=None): + self.config, self.jump_id = config, jump_id + if jump_id not in config['hosts'] or config['hosts'][jump_id].get('role') != 'white_jump': + raise ValueError('Explicit registered white jump required') + if client_factory is None: + known=Path(config['known_hosts']) + if not known.is_absolute() or known.is_symlink() or not known.is_file() or known.stat().st_mode & 0o022:raise ValueError('Explicit trusted known-hosts file required') + if any(not re.fullmatch('SHA256:[A-Za-z0-9+/]{43}=?',s.get('host_key_sha256','')) for s in config['hosts'].values()):raise ValueError('Approved explicit host fingerprints required') + self.factory = client_factory + self.local = threading.local() + self.clients = [] + self.lock = threading.Lock() + + def get(self, rid): + sessions = getattr(self.local, 'sessions', None) + if sessions is None: + sessions = self.local.sessions = {} + if rid in sessions: + c = sessions[rid] + if c.get_transport() and c.get_transport().is_active(): + return c + # Do not transparently retry operations after a lost connection. + raise RuntimeError('SSH session lost; probe remote outcome on next run') + spec = self.config['hosts'][rid] + jump = self.get(self.jump_id) if rid != self.jump_id else None + if self.factory: + client = self.factory(rid, spec, jump) + else: + import paramiko + client = paramiko.SSHClient() + known=Path(self.config['known_hosts']) + if not known.is_absolute() or known.is_symlink() or not known.is_file() or known.stat().st_mode & 0o022:raise ValueError('Trusted non-writable known-hosts reference required') + entries=paramiko.HostKeys(str(known)) + hostname=spec['host'] if spec.get('port',22)==22 else '['+spec['host']+']:'+str(spec['port']) + approved=[] + for key in (entries.lookup(hostname) or {}).values(): + actual='SHA256:'+base64.b64encode(hashlib.sha256(key.asbytes()).digest()).decode().rstrip('=') + if actual==spec['host_key_sha256'].rstrip('='):approved.append(key) + if not approved:raise ValueError('Approved known-host key absent; no credentials transmitted') + for key in approved:client.get_host_keys().add(hostname,key.get_name(),key) + client.set_missing_host_key_policy(paramiko.RejectPolicy()) + auth = credentials(spec) + if set(auth) - {'username', 'password', 'key_filename', 'passphrase'}: + raise ValueError('Unexpected SSH credential fields') + if auth.get('username',spec['username'])!=spec['username']:raise ValueError('Credential username differs from registered host') + if auth.get('key_filename'): + key=Path(auth['key_filename']) + if not key.is_absolute() or not key.is_file() or key.stat().st_mode & 0o077:raise ValueError('Private absolute SSH key required') + sock = jump.get_transport().open_channel('direct-tcpip', (spec['host'], spec.get('port', 22)), ('127.0.0.1', 0)) if jump else None + try: + client.connect(spec['host'], port=spec.get('port', 22), username=spec['username'], + password=auth.get('password'), key_filename=auth.get('key_filename'), + passphrase=auth.get('passphrase'), sock=sock, allow_agent=False, + look_for_keys=False, timeout=20, banner_timeout=20, auth_timeout=20) + except Exception: + client.close() + raise RuntimeError('SSH connect failed; inspect registered route/host key/credential reference') from None + finally: + auth.clear() + sessions[rid] = client + with self.lock: + self.clients.append(client) + return client + + def close(self): + for client in reversed(self.clients): + client.close() + self.clients.clear() diff --git a/tests/node-onboarding/AcceptanceUserIsolatedTest.php b/tests/node-onboarding/AcceptanceUserIsolatedTest.php new file mode 100644 index 0000000..fa49600 --- /dev/null +++ b/tests/node-onboarding/AcceptanceUserIsolatedTest.php @@ -0,0 +1,66 @@ +instance('app',$app); + $app->instance('config',new Illuminate\Config\Repository(['cache'=>['default'=>'array','stores'=>['array'=>['driver'=>'array']]],'app'=>['key'=>'offline-test']])); + $events=new Illuminate\Events\Dispatcher($app);$app->instance('events',$events); + $this->db=new Capsule($app);$this->db->addConnection(['driver'=>'sqlite','database'=>':memory:']);$this->db->setEventDispatcher($events);$this->db->setAsGlobal();$this->db->bootEloquent();$app->instance('db',$this->db->getDatabaseManager());$app->singleton('cache',fn($a)=>new Illuminate\Cache\CacheManager($a)); + $s=$this->db->schema(); + $s->create('v2_user',function($t){$t->increments('id');$t->string('email')->unique();foreach(['uuid','token','password'] as $x)$t->string($x);$t->integer('group_id')->nullable();foreach(['u','d'] as $x)$t->integer($x)->default(0);$t->integer('transfer_enable');$t->integer('expired_at')->nullable();foreach(['banned','is_admin','remind_expire','remind_traffic'] as $x)$t->boolean($x)->default(0);foreach(['device_limit','speed_limit'] as $x)$t->integer($x)->nullable();$t->integer('created_at')->nullable();$t->integer('updated_at')->nullable();}); + $s->create('v2_server_group',function($t){$t->increments('id');$t->string('name');});Capsule::table('v2_server_group')->insert([['id'=>1,'name'=>'formal'],['id'=>2,'name'=>'other']]); + $s->create('v2_server',function($t){$t->increments('id');foreach(['name','type','host','port'] as $x)$t->string($x);foreach(['group_ids','protocol_settings'] as $x)$t->text($x);$t->boolean('enabled')->default(1);$t->boolean('show')->default(1);$t->integer('sort')->default(1);foreach(['u','d'] as $x)$t->integer($x)->default(0);$t->integer('transfer_enable')->nullable();$t->float('rate')->default(1);$t->timestamps();}); + $this->node();$this->dir=sys_get_temp_dir().'/dpl-acceptance-'.bin2hex(random_bytes(8));mkdir($this->dir,0700); + } + protected function tearDown():void{foreach(glob($this->dir.'/*') as $p)unlink($p);rmdir($this->dir);$this->db->getDatabaseManager()->disconnect();} + private function node():void {$i=Server::count()+1;Server::create(['name'=>'node'.$i,'type'=>'anytls','host'=>'relay.example.invalid','port'=>(string)(47000+$i),'group_ids'=>['1'],'protocol_settings'=>['tls'=>['server_name'=>'n'.$i.'.example.invalid','allow_insecure'=>false]]]);} + private function user(array $more=[]):User{return User::create(array_replace(['email'=>'existing@example.invalid','uuid'=>'existing-uuid','token'=>'existing-token','password'=>'unused','group_id'=>1,'transfer_enable'=>1000000,'expired_at'=>null],$more));} + private function reject(callable $f):void{$caught=false;try{$f();}catch(RuntimeException $e){$caught=true;}self::assertTrue($caught);} + public function testPreviewDoesNotCreateUserOrLease():void { + self::assertSame('would_create',Acceptance::prepare($this->dir,1,false)['state']);self::assertSame(0,User::count());self::assertFileDoesNotExist($this->dir.'/acceptance-user.json'); + } + public function testCreatesBoundedNonAdminAndUsesDynamicActualNodeSet():void { + $r=Acceptance::prepare($this->dir,1,true);$u=User::findOrFail($r['id']);self::assertTrue($r['owned']);self::assertFalse($u->is_admin);self::assertSame(104857600,(int)$u->transfer_enable);self::assertLessThanOrEqual(time()+3600,(int)$u->expired_at);self::assertGreaterThan(time()+3500,(int)$u->expired_at); + self::assertCount(1,Acceptance::source($this->dir,1)['nodes']);$this->node();self::assertCount(2,Acceptance::source($this->dir,1)['nodes']);self::assertSame($u->id,Acceptance::prepare($this->dir,1,true)['id']);self::assertSame(1,User::count()); + Acceptance::cleanup($this->dir,true);self::assertTrue($u->fresh()->banned);self::assertTrue(App\Services\ServerService::getAvailableUsers(Server::first())->isEmpty());self::assertArrayNotHasKey('token',Acceptance::source($this->dir,1,true));$this->reject(fn()=>Acceptance::source($this->dir,1)); + } + public function testExistingEligibleAdminIsUsedWithoutPrivilegeOrPasswordChange():void { + $u=$this->user(['is_admin'=>1]);$before=$u->fresh()->toArray();$r=Acceptance::prepare($this->dir,1,true);self::assertFalse($r['owned']);self::assertSame($u->id,$r['id']);Acceptance::source($this->dir,1);Acceptance::cleanup($this->dir,true);self::assertSame($before,$u->fresh()->toArray()); + } + public function testExpiredExhaustedBannedWrongGroupAreNotReused():void { + foreach([['expired_at'=>time()-1],['u'=>1000000],['banned'=>1],['group_id'=>2]] as $i=>$change)$this->user($change+['email'=>'existing'.$i.'@example.invalid']); + $r=Acceptance::prepare($this->dir,1,true);self::assertTrue($r['owned']);self::assertSame(5,User::count()); + } + public function testLostCommitResponseRecoversSameUserAndPreparedRollbackRecoversSameIdentity():void { + $r=Acceptance::prepare($this->dir,1,true);$p=$this->dir.'/acceptance-user.json';$j=DplOnboarding\readJson($p);$j['state']='prepared';unset($j['id']);DplOnboarding\writeJson($p,$j); + self::assertSame($r['id'],Acceptance::prepare($this->dir,1,true)['id']);self::assertSame(1,User::count()); + User::find($r['id'])->delete();DplOnboarding\writeJson($p,$j);$again=Acceptance::prepare($this->dir,1,true);self::assertSame(1,User::count());self::assertSame($j['uuid'],User::find($again['id'])->uuid); + } + public function testClosedOwnedUserNeverReactivatedOnResume():void { + $old=Acceptance::prepare($this->dir,1,true)['id'];Acceptance::cleanup($this->dir,true);Acceptance::cleanup($this->dir,true); + $new=Acceptance::prepare($this->dir,1,true)['id'];self::assertNotSame($old,$new);self::assertTrue(User::find($old)->banned);self::assertFalse(User::find($new)->banned);self::assertFileExists($this->dir.'/acceptance-closed-'.$old.'.json');Acceptance::cleanup($this->dir,true); + } + public function testCleanupRefusesDriftAndNeverBansSomeoneElsesAccount():void { + $id=Acceptance::prepare($this->dir,1,true)['id'];$u=User::find($id);$u->is_admin=true;$u->saveQuietly();$this->reject(fn()=>Acceptance::cleanup($this->dir,true));self::assertFalse($u->fresh()->banned); + } + public function testEffectivePluginExclusionBlocksAcceptanceSource():void { + Acceptance::prepare($this->dir,1,true); + App\Services\Plugin\HookManager::registerFilter('server.users.get',fn($users)=>collect()); + $this->reject(fn()=>Acceptance::source($this->dir,1));Acceptance::cleanup($this->dir,true); + } + public function testFailureFinallyCleanupAndMissingGroup():void { + $this->reject(fn()=>Acceptance::prepare($this->dir,99,true));$id=Acceptance::prepare($this->dir,1,true)['id']; + try{throw new RuntimeException('Injected client failure');}catch(RuntimeException $e){}finally{Acceptance::cleanup($this->dir,true);} + self::assertTrue(User::find($id)->banned); + } +} diff --git a/tests/node-onboarding/test_runtime.py b/tests/node-onboarding/test_runtime.py new file mode 100644 index 0000000..d64a00f --- /dev/null +++ b/tests/node-onboarding/test_runtime.py @@ -0,0 +1,312 @@ +"""Offline adversarial evidence/transport/resume tests; no network or Laravel env.""" +import copy +import hashlib +import importlib.util +import json +from pathlib import Path +import sys +import tempfile +import threading +import time +import unittest +from unittest.mock import patch + +SCRIPTS=Path(__file__).resolve().parents[2]/'scripts/node-onboarding' +sys.path.insert(0,str(SCRIPTS)) +from acceptance import audit,digest,snapshot,service_change,provider_state +from ssh_pool import Pool,private_json,credentials +import batch + +class EvidenceTests(unittest.TestCase): + def fixture(self,count): + nodes=[dict(id=i,type='anytls',tag='node-'+str(i),relay='relay.example.invalid',relay_port=47000+i,sni=f'n{i}.example.invalid',expected_exit=f'192.0.2.{i}',country='JP') for i in range(1,count+1)] + report={'state':'live_verified','verified_at':1000,'results':[dict(tag=n['tag'],relay=n['relay'],relay_port=n['relay_port'],sni=n['sni'],exit_ip=n['expected_exit'],country=n['country'],live=True,strict_tls=True) for n in nodes]} + qas={'captured_at':1000,'plan_sha256':'plan','users':{'new':dict(id=9,http_status=403,banned=True,excluded=True)}} + return nodes,report,qas + def check(self,nodes,report,qas,after=None,**bindings): + raw=json.dumps(report).encode();qraw=json.dumps(qas).encode() + binding=dict(plan_sha256='plan',snapshot_sha256=digest(snapshot(nodes)),report_sha256=hashlib.sha256(raw).hexdigest(),qa_rejections_sha256=hashlib.sha256(qraw).hexdigest(),verified_at=1000) + binding.update(bindings) + return audit(nodes,nodes if after is None else after,raw,qraw,binding,'plan',{'new':9},now=1100) + def test_arbitrary_node_counts_and_reordering(self): + for count in [1,2,7,23]: + nodes,report,qas=self.fixture(count) + self.assertEqual(self.check(nodes,report,qas,after=list(reversed(nodes)))['nodes'],count) + def test_topology_change_including_old_node_is_blocked(self): + nodes,r,q=self.fixture(7) + for after in [nodes[:-1],nodes+self.fixture(8)[0][-1:],list(reversed(nodes[:-1]))]: + with self.assertRaises(ValueError):self.check(nodes,r,q,after=after) + altered=copy.deepcopy(nodes);altered[0]['expected_exit']='198.51.100.9' + with self.assertRaises(ValueError):self.check(nodes,r,q,after=altered) + def test_missing_extra_duplicate_or_insecure_results_fail(self): + nodes,r,q=self.fixture(7) + for results in [r['results'][:-1],r['results']+[r['results'][0]],r['results'][1:]+[r['results'][1]]]: + with self.assertRaises(ValueError):self.check(nodes,dict(r,results=results),q) + for field,value in [('exit_ip','198.51.100.1'),('country','US'),('strict_tls',False),('live',False),('sni','wrong.example.invalid')]: + wrong=copy.deepcopy(r);wrong['results'][0][field]=value + with self.assertRaises(ValueError):self.check(nodes,wrong,q) + def test_qa_identity_status_revocation_and_hashes_are_bound(self): + nodes,r,q=self.fixture(2) + for field,value in [('id',10),('http_status',200),('banned',False),('excluded',False)]: + wrong=copy.deepcopy(q);wrong['users']['new'][field]=value + with self.assertRaises(ValueError):self.check(nodes,r,wrong) + for field in ['snapshot_sha256','plan_sha256','report_sha256','qa_rejections_sha256']: + with self.assertRaises(ValueError):self.check(nodes,r,q,**{field:'wrong'}) + for stamp in [-1,1200,None,'1000']: + with self.assertRaises(ValueError):self.check(nodes,dict(r,verified_at=stamp),q) + with self.assertRaises(ValueError):self.check(nodes,r,dict(q,captured_at=stamp)) + def test_provider_observations_are_exact_and_not_language_or_substring_dependent(self): + node=dict(key='new',tls_domain='a.example.invalid',relay_host='relay.example.invalid',relay_port=47001,node_address='192.0.2.1',listen_port=24443) + observed={'captured_at':1000,'dns':[{'type':'A','name':node['tls_domain'],'address':node['node_address'],'proxy':False}],'relays':[{'name':'dpl-new-anytls','host':node['relay_host'],'port':47001,'target':node['node_address'],'listen_port':24443,'proxy_protocol_in':False,'proxy_protocol_out':False}]} + self.assertTrue(provider_state(observed,node,1100)['dns']) + for section,field,value in [('dns','proxy',True),('dns','address','192.0.2.11'),('relays','proxy_protocol_in',True),('relays','target','192.0.2.11')]: + wrong=copy.deepcopy(observed);wrong[section][0][field]=value + with self.assertRaises(ValueError):provider_state(wrong,node,1100) + self.assertEqual(provider_state(dict(observed,dns=[],relays=[]),node,1100),{'dns':[],'relays':[]}) + with self.assertRaises(ValueError):provider_state(observed,node,2000) + + def test_registry_and_service_changes_do_not_use_pid_alone(self): + for values in [[],[dict(self.fixture(1)[0][0],type='vless')],self.fixture(1)[0]*2]: + with self.assertRaises(ValueError):snapshot(values) + before=dict(unit_sha256='a',binary_sha256='b',config_sha256='c',listeners=['*:443'],active=True,pid=1,started='yesterday',restarts=0) + self.assertEqual(service_change(before,before),'ready') + self.assertEqual(service_change(before,dict(before,pid=2)),'investigate_restart') + self.assertEqual(service_change(before,dict(before,config_sha256='changed')),'conflict') + self.assertEqual(service_change(before,dict(before,active=False)),'conflict') + self.assertEqual(service_change(before,dict(before,binary_sha256=None)),'unknown') + +class TransportTests(unittest.TestCase): + def test_reuse_and_every_nonwhite_host_has_jump(self): + calls=[] + class Client: + active=True + def get_transport(self):return self + def is_active(self):return self.active + def close(self):self.active=False + def factory(rid,spec,jump):calls.append((rid,jump));return Client() + config={'hosts':{'white':{'role':'white_jump'},'panel':{},'node':{}}} + pool=Pool(config,'white',factory) + panel=pool.get('panel');self.assertIs(panel,pool.get('panel')) + node=pool.get('node');self.assertIs(calls[1][1],calls[2][1]);self.assertIsNone(calls[0][1]);self.assertEqual(len(calls),3) + panel.active=False + with self.assertRaises(RuntimeError):pool.get('panel') + self.assertEqual(len(calls),3) # Never automatically replay after disconnect. + pool.close();self.assertFalse(node.active) + def test_sessions_are_not_shared_unsafely_between_workers(self): + barrier=threading.Barrier(2);made=[] + class Client: + def get_transport(self):return self + def is_active(self):return True + def close(self):pass + pool=Pool({'hosts':{'w':{'role':'white_jump'},'n':{}}},'w',lambda *args:Client()) + def worker(): + n=pool.get('n');barrier.wait();made.append(n);self.assertIs(n,pool.get('n')) + threads=[threading.Thread(target=worker) for _ in range(2)] + for t in threads:t.start() + for t in threads:t.join(timeout=5) + self.assertEqual(len(made),2);self.assertIsNot(made[0],made[1]);pool.close() + def test_existing_codocs_credentials_are_read_without_copy_or_rewrite(self): + with tempfile.TemporaryDirectory() as d: + p=Path(d)/'vps.md';body='# Fixture\n密码：test-only-phrase\n';p.write_text(body);p.chmod(0o600) + self.assertEqual(credentials({'credential_ref':str(p),'credential_format':'codocs-password'}),{'password':'test-only-phrase'}) + self.assertEqual(p.read_text(),body);self.assertEqual(len(list(Path(d).iterdir())),1) + p.write_text(body+'密码：ambiguous\n') + with self.assertRaises(ValueError):credentials({'credential_ref':str(p),'credential_format':'codocs-password'}) + def test_private_config_and_explicit_jump_enforced(self): + with self.assertRaises(ValueError):Pool({'hosts':{'n':{}}},'w') + with self.assertRaises(ValueError):Pool({'hosts':{'w':{}}},'w') + with tempfile.TemporaryDirectory() as d: + p=Path(d)/'auth.json';p.write_text('{}');p.chmod(0o644) + with self.assertRaises(ValueError):private_json(p) + p.chmod(0o600);self.assertEqual(private_json(p),{}) + link=Path(d)/'link';link.symlink_to(p) + with self.assertRaises(ValueError):private_json(link) + +class ResumeTests(unittest.TestCase): + def setUp(self): + batch.STOP.clear();self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name) + spec=importlib.util.spec_from_file_location('batch_fixtures',Path(__file__).with_name('test_batch.py'));m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m) + fixture=m.BatchTests();fixture.setUp();self.fixture=fixture;self.plan=fixture.plan + self.calls=[];self.generation='a';self.finished=False + owner=self + class Runner(batch.Coordinator): + def call(self,n,phase,action): + owner.calls.append((phase,action)) + r=dict(key=n['key'],phase=phase,request_sha256='irrelevant-in-unit-fixture',status='waiting' if phase=='audit' and not owner.finished else 'ready',valid_until=time.time()+180) + if phase=='preflight':r['resume_dependencies']={k:hashlib.sha256((owner.generation if k=='provision' else k).encode()).hexdigest() for k in ['stage','dns','provision','relay','verify']} + return r + self.Runner=Runner + def tearDown(self):self.fixture.tearDown();self.tmp.cleanup();batch.STOP.clear() + def run_batch(self):return self.Runner(self.plan,self.root/'ledger',jobs=1).run() + def test_fresh_proofs_skip_only_completed_unchanged_phases(self): + self.assertFalse(self.run_batch());self.calls.clear();self.finished=True + self.assertTrue(self.run_batch());self.assertEqual(self.calls,[('preflight','probe'),('publish','probe'),('audit','probe')]) + def test_dependency_change_rechecks_downstream_and_not_unrelated_stage(self): + self.run_batch();self.calls.clear();self.generation='b';self.finished=True;self.run_batch() + self.assertEqual([p for p,a in self.calls],['preflight','provision','relay','verify','publish','audit']) + def test_expired_and_unknown_proofs_never_skip(self): + self.run_batch();runner=self.Runner(self.plan,self.root/'ledger',jobs=1) + with runner.ledger.db() as c:c.execute('UPDATE evidence SET expires=0 WHERE phase="dns"') + runner.ledger.mark(self.plan['batch_id'],self.plan['nodes'][0]['key'],'stage','unknown') + self.calls.clear();self.finished=True;self.run_batch() + self.assertIn(('stage','probe'),self.calls);self.assertIn(('dns','probe'),self.calls) + self.assertTrue((self.root/'ledger/events.jsonl').is_file()) + + +class SiteAdapterTests(unittest.TestCase): + def test_preserved_service_review_binds_original_and_exact_observation(self): + from site_adapter import SiteAdapter + adapter=object.__new__(SiteAdapter);adapter.ctx={'preserved_service':dict(unit='XrayR',binary='/bin/xrayr',config='/etc/xrayr.yml',port=443)} + current=dict(unit_sha256='u',binary_sha256='b',config_sha256='c',listeners=['*:443'],active=True,pid=1,started='a',restarts=0) + files={};read=lambda client,path:files.get(path) + def write(client,path,value):files[path]=value.encode() if isinstance(value,str) else value + def execute(client,cmd):return json.dumps(current) if cmd.startswith('python3') else '' + adapter.guard_service(None,'/backup',read,write,execute) + original=copy.deepcopy(current);current.update(pid=2,started='b',restarts=1) + with self.assertRaisesRegex(RuntimeError,'investigate_'):adapter.guard_service(None,'/backup',read,write,execute) + files['/backup/preserved-service-review.json']=json.dumps(dict(before_sha256=digest(original),after_sha256=digest(current),reason='Independently reviewed lifecycle panic',evidence_sha256='e'*64)).encode() + adapter.guard_service(None,'/backup',read,write,execute) + current['config_sha256']='changed' + with self.assertRaisesRegex(RuntimeError,'conflict'):adapter.guard_service(None,'/backup',read,write,execute) + self.assertEqual(json.loads(files['/backup/preserved-service-original.json']),original) + def test_generated_remote_python_and_group_independent_php_compile(self): + import ast + import subprocess + import io + from site_adapter import SiteAdapter + tree=ast.parse((SCRIPTS/'site_adapter.py').read_text()) + for method in [n for n in ast.walk(tree) if isinstance(n,ast.FunctionDef) and n.name in ('formal','guard_service','certificate_gate')]: + for c in ast.walk(method): + if isinstance(c,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='script' for t in c.targets): + program=eval(compile(ast.Expression(c.value),'script-builder','eval'),{'repr':repr,'spec':{},'ctx':{'subscription_url':'https://sub.example.invalid'},'value':{'token':'test-only'}}) + compile(program,'remote-'+method.name,'exec') + with tempfile.TemporaryDirectory() as d: + a=object.__new__(SiteAdapter);a.plan={'context':dict(panel_vps='panel',gateway_vps='gateway',node_root='/node',ops_root='/ops',app_root='/app',batch_backup_root='/batch',node_backup_root='/backups',qa_root='/qa',gateway_containers=['gateway-a','gateway-b'])};a.ctx=a.plan['context'];a.directory=Path(d) + php_sources=[];calls=[] + class Channel: + def shutdown_write(self):pass + def recv_exit_status(self):return 0 + class In: + channel=Channel() + def write(self,body):php_sources.append(body) + class Out: + channel=Channel() + def read(self): + body=php_sources[-1] + if '$base=json_decode' in body:return b'true' + return json.dumps({'manifest':{},'node_online':True,'groups':['7'],'qa_banned':True,'u':1,'d':1}).encode() + class DockerOut: + channel=Channel() + def read(self):return json.dumps([dict(Name='/gateway-a',Id='id-a',State={'StartedAt':'then'},RestartCount=0)]).encode() + class Sftp: + def __enter__(self):return self + def __exit__(self,*args):pass + def open(self,path,mode):return io.BytesIO(json.dumps({'/gateway-a':{'id':'id-a','started':'then','restarts':0}}).encode()) + class Remote: + def exec_command(self,cmd,timeout=0): + calls.append(cmd) + if cmd.startswith('docker inspect'): + if cmd!='docker inspect gateway-a gateway-b':raise RuntimeError('Invalid gateway inventory command') + return In(),DockerOut(),io.BytesIO() + return In(),Out(),io.BytesIO() + def open_sftp(self):return Sftp() + class Pool: + def get(self,rid):return Remote() + a.pool=Pool();a.guard_service=lambda *args:None;a.formal=lambda *args:None + n=dict(key='sg-new',request_sha256='a'*64,formal_group_id=7,vps_id='node') + result=a.call(n,'audit','probe');self.assertEqual(result['status'],'ready');self.assertGreaterEqual(len(php_sources),2) + for php in php_sources: + result=subprocess.run(['php','-l'],input=php,text=True,capture_output=True) + self.assertEqual(result.returncode,0,result.stderr) + def test_pool_real_connect_interface_keeps_strict_keys_and_jump_sock(self): + import types + recorded=[] + class FakeClient: + def load_host_keys(self,path):recorded.append(('keys',path)) + def set_missing_host_key_policy(self,p):recorded.append(('reject',p)) + def get_host_keys(self):return self + def add(self,*args):recorded.append(('approved-key',args)) + def connect(self,host,**kw):recorded.append(('connect',host,kw)) + def get_transport(self):return self + def is_active(self):return True + def open_channel(self,kind,target,origin):recorded.append(('jump',kind,target));return 'through-white' + def close(self):pass + class Key: + def asbytes(self):return b'approved-public-key-fixture' + def get_name(self):return 'ssh-ed25519' + class Keys: + def __init__(self,path):pass + def lookup(self,name):return {'ssh-ed25519':Key()} + module=types.SimpleNamespace(SSHClient=FakeClient,RejectPolicy=lambda:'reject-unknown',HostKeys=Keys) + pin='SHA256:'+__import__('base64').b64encode(hashlib.sha256(Key().asbytes()).digest()).decode().rstrip('=') + with tempfile.TemporaryDirectory() as d,patch.dict(sys.modules,{'paramiko':module}): + keys=Path(d)/'known_hosts';keys.write_text('public fixture');keys.chmod(0o600) + credentials=Path(d)/'credentials.json';credentials.write_text(json.dumps({'password':'test-only-fixture'}));credentials.chmod(0o600) + hosts={'w':dict(role='white_jump',host='192.0.2.1',username='root',credential_ref=str(credentials)),'n':dict(host='192.0.2.2',username='root',credential_ref=str(credentials))} + + for host in hosts.values():host['host_key_sha256']=pin + p=Pool(dict(hosts=hosts,known_hosts=str(keys)),'w');p.get('n');p.get('n');p.close() + before_connects=len([x for x in recorded if x[0]=='connect']) + hosts['w']['host_key_sha256']='SHA256:'+'b'*43 + blocked=Pool(dict(hosts=hosts,known_hosts=str(keys)),'w') + with self.assertRaises(ValueError):blocked.get('n') + self.assertEqual(len([x for x in recorded if x[0]=='connect']),before_connects) + blocked.close() + connections=[x for x in recorded if x[0]=='connect'];self.assertEqual(len(connections),2);self.assertIsNone(connections[0][2]['sock']);self.assertEqual(connections[1][2]['sock'],'through-white') + self.assertEqual(len([x for x in recorded if x[0]=='reject']),2) + for _,_,kw in connections:self.assertFalse(kw['allow_agent']);self.assertFalse(kw['look_for_keys']) + + def test_configure_is_local_only_and_pins_code_context_and_interpreter(self): + import types + spec=importlib.util.spec_from_file_location('configure_site',SCRIPTS/'configure-site.py');module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module) + with tempfile.TemporaryDirectory() as d: + root=Path(d);ssh=root/'ssh.json';registry=root/'exits.json';(root/'known_hosts').write_text('public fixture');(root/'known_hosts').chmod(0o600) + hosts={'VPS-001':dict(role='white_jump',host='192.0.2.1',username='root'),'VPS-002':dict(host='192.0.2.2',username='root'),'VPS-003':dict(host='192.0.2.3',username='root'),'VPS-011':dict(host='192.0.2.11',username='root')} + for host in hosts.values():host['host_key_sha256']='SHA256:'+'a'*43 + ssh.write_text(json.dumps({'hosts':hosts,'known_hosts':str(root/'known_hosts')}));ssh.chmod(0o600);registry.write_text('{}');registry.chmod(0o600) + fixture_spec=importlib.util.spec_from_file_location('base_fixture',Path(__file__).with_name('test_batch.py'));f=importlib.util.module_from_spec(fixture_spec);fixture_spec.loader.exec_module(f) + original=f.BatchTests();original.setUp() + try: + base=original.plan;node=base['nodes'][0];node['node_address']='192.0.2.11';node['expected_exit']='192.0.2.11';base['context'].update(jump_vps='VPS-001',panel_vps='VPS-002',gateway_vps='VPS-003');registry.write_text(json.dumps({node['tls_domain']:{'expected_exit':node['expected_exit'],'country':node['country']}})) + site=dict(client_sha256='d'*64,ops_root='/ops',node_root='/node',batch_backup_root='/backup/batch',node_backup_root='/backup/nodes',qa_root='/qa/batch',provider_snapshot=str(root/'provider.json'),exit_registry=str(registry),ssh_config=str(ssh),preserved_service=dict(unit='XrayR',binary='/bin/xrayr',config='/etc/xrayr.yml',port=443),formal_socks_port=19999,gateway_containers=['gateway-a']) + with patch.dict(sys.modules,{'paramiko':types.SimpleNamespace(__version__='5.0.0')}): + plan=module.configure(base,site);self.assertEqual(plan['runtime']['kind'],'ssh-site-v1');self.assertFalse((root/'ledger').exists()) + runner=batch.Coordinator(plan,root/'ledger',jobs=1);runner.runtime.close();runner.check_runtime() + ssh.write_text(ssh.read_text()+'\n') + with self.assertRaises(ValueError):runner.check_runtime() + finally:original.tearDown() + +class PackagingTests(unittest.TestCase): + def test_committed_flat_snapshot_preserves_worktree_index_and_base(self): + import subprocess + spec=importlib.util.spec_from_file_location('bundle_ops',SCRIPTS/'bundle-ops.py');module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module) + with tempfile.TemporaryDirectory() as d: + root=Path(d);repo=root/'repo';repo.mkdir() + def git(*args):return subprocess.check_output(['git','-C',str(repo),*args],stderr=subprocess.DEVNULL) + git('init');git('config','user.name','Offline Test');git('config','user.email','test@example.invalid');git('config','commit.gpgsign','false') + (repo/'base-only.txt').write_text('preserved operations base');git('add','.');git('commit','-m','base');base=git('rev-parse','HEAD').decode().strip() + source_dir=repo/'scripts/node-onboarding';source_dir.mkdir(parents=True) + for name in module.FILES:(source_dir/name).write_text('committed '+name) + git('add','.');git('commit','-m','source');source=git('rev-parse','HEAD').decode().strip() + (repo/'base-only.txt').write_text('unrelated user modification');(repo/'unrelated.txt').write_text('staged unrelated');git('add','unrelated.txt') + index_before=git('diff','--cached','--binary');status_before=git('status','--porcelain') + output=root/'operations.bundle';result=module.build(repo,source,base,'codex/offline-ops',output) + self.assertTrue(output.exists());self.assertEqual(git('diff','--cached','--binary'),index_before);self.assertEqual(git('status','--porcelain'),status_before) + self.assertEqual(git('show',result['commit']+':panel-acceptance.php'),b'committed panel-acceptance.php') + self.assertEqual(git('show',result['commit']+':base-only.txt'),b'preserved operations base') + self.assertEqual(git('rev-parse','HEAD').decode().strip(),source) + git('bundle','verify',str(output)) + with self.assertRaises(ValueError):module.build(repo,source,base,'codex/offline-ops',root/'second.bundle') + def test_event_stream_preserves_final_result_and_avoids_status_polling(self): + import subprocess + spec=importlib.util.spec_from_file_location('event_fixture',Path(__file__).with_name('test_batch.py'));f=importlib.util.module_from_spec(spec);spec.loader.exec_module(f) + test=f.BatchTests();test.setUp() + try: + result=subprocess.run(test.argv()+['--events'],capture_output=True,text=True,timeout=30) + self.assertEqual(result.returncode,0,result.stderr) + lines=[json.loads(x) for x in result.stdout.splitlines()] + self.assertTrue(lines[-1]['complete']);self.assertTrue(any(x.get('phase')=='provision' for x in lines[:-1])) + for x in lines[:-1]:self.assertEqual(set(x),{'batch','node','phase','status','time'}) + finally:test.tearDown() + +if __name__=='__main__':unittest.main()