fix(nodes): validate QA scripts and normalized formal snapshots

docs/maintenance/STATUS.md

@@ -2,6 +2,12 @@
 
 文档核对日期：2026-10-11。以下区分本会话检查结果与用户验收确认，不承诺服务器一直保持此状态。
 
+## 2026-10-17 后续五节点接入（进行中）
+
+本批授权新增5个节点，复用一个核查验收助手。准备阶段独立审查发现并修复正式适配器的两处调用链阻断：QA订阅远端Python脚本缩进错误、标准化快照未保留协议类型导致最终audit无法重复校验。Python全套34项通过，新增覆盖实际发送的QA脚本编译，以及正式source→snapshot→audit→缓存audit链路；这些是离线回归，不表示本批线上接入或验收完成。
+
+候选jp06/07/08、us03现场只读核查通过；us02因SSH主机密钥与既有信任不符停止，用户批准备用us04精确首次指纹后核查通过。5台原XrayR均在运行，新接入端口24443空闲；Debian11/10的Git依赖按各自发行版准备，不沿用硬编码bullseye源。本批账本与现场证据在忽略目录`codex/output/node-batch-five-20261017/`。正式开放、动态全量订阅、QA清理、旧资源回归及资源文档同步尚待完成。
+
 ## 2026-10-17 批量接入流程优化（本地验证）
 
 已实现正式`ssh-site-v1`适配器及`configure-site.py`入口，去除临时handler中的固定日期、日本、组1和节点数量。正式验收从真实账号授权集合生成快照，完整校验新旧节点、预期出口、严格TLS、原始证据摘要、QA身份/403/禁用/下发排除及时效；集合变化停止。没有有效正式组账号时自动创建100MiB/1小时/1设备非管理员测试用户，journal恢复同一身份，正常成功/失败/等待均清理本任务用户，关闭后不复活；异常中断依靠日志恢复和时限兜底。已批准指纹匹配后才读取凭据，SSH始终经白色机，按线程复用会话、按阶段延迟连接，不自动重放断线命令。

scripts/node-onboarding/acceptance.py

@@ -13,7 +13,7 @@ def digest(value):
 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')
+    fields = ('id', 'type', '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):

scripts/node-onboarding/site_adapter.py

@@ -297,16 +297,16 @@ for attempt in range(3 if source['owned'] else 1):
                 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')
+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')}))
+'''
             return gateway_py(gateway,'Q='+repr(Q)+';SUB='+repr(ctx['subscription_url'])+'\n'+script)
         def phase_impl(panel,gateway,node):
             if a.phase=='preflight':

tests/node-onboarding/test_runtime.py

@@ -32,6 +32,12 @@ class EvidenceTests(unittest.TestCase):
         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_snapshot_can_be_revalidated_by_formal_audit(self):
+        nodes,report,qas=self.fixture(2)
+        normalized=snapshot(nodes)
+        self.assertEqual(snapshot(normalized),normalized)
+        self.assertTrue(all(n['type']=='anytls' for n in normalized))
+        self.assertEqual(self.check(normalized,report,qas)['nodes'],2)
     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]))]:
@@ -194,11 +200,16 @@ class SiteAdapterTests(unittest.TestCase):
                 def chmod(self,*args):pass
                 def posix_rename(self,old,new):self.files[new]=self.files.pop(old)
             state=dict(manifest={'test_user_id':9},node_online=True,groups=['2'],qa_banned=False,u=12,d=34)
+            qa_programs=[]
             class Remote:
                 def __init__(self,kind):self.kind=kind;self.files={}
                 def open_sftp(self):return Sftp(self.files)
                 def exec_command(self,cmd,timeout=0):
                     if cmd.endswith(' && php'):return In(),Out(json.dumps(state).encode()),Out()
+                    if cmd.startswith("python3 - <<'DPLPY'\n"):
+                        program=cmd.split('\n',1)[1].rsplit('\nDPLPY',1)[0]
+                        compile(program,'actual-fetch-qa-remote','exec')
+                        qa_programs.append(program)
                     if 'verify-node.py' in cmd:self.files['/qa/new/qa-report.json']=raw
                     return In(),Out(b'{}'),Out()
             remotes={key:Remote(key) for key in ['panel','gateway','node']}
@@ -208,9 +219,63 @@ class SiteAdapterTests(unittest.TestCase):
             adapter.pool=Pool()
             node=dict(key='new',request_sha256='request',formal_group_id=1,vps_id='node',socks_port=19011,expected_exit='192.0.2.11',country='JP')
             result=adapter.call(node,'verify','probe');self.assertEqual(result['status'],'ready')
+            self.assertEqual(len(qa_programs),1)
+            self.assertIn('urllib.request.urlopen',qa_programs[0])
             exported=json.loads((Path(d)/'new-evidence.json').read_text());self.assertEqual(exported['report_sha256'],hashlib.sha256(raw).hexdigest());self.assertEqual(exported['panel']['d'],34)
             self.assertEqual(remotes['panel'].files['/backups/new/qa-report.json'],raw)
 
+    def test_formal_source_snapshot_audit_and_cached_audit_call_chain(self):
+        import contextlib
+        import io
+        from site_adapter import SiteAdapter
+        adapter=object.__new__(SiteAdapter)
+        ctx=dict(gateway_vps='gateway',qa_root='/qa',ops_root='/ops',subscription_url='https://sub.example.invalid',client_binary='/client',formal_socks_port=19999,node_backup_root='/backups',batch_backup_root='/batch')
+        candidate=dict(key='new',formal_group_id=7,relay_host='relay.example.invalid',relay_port=47002,tls_domain='n2.example.invalid')
+        adapter.plan={'context':ctx,'nodes':[candidate]};adapter.ctx=ctx
+        adapter.directory=Path('/private/tmp/formal-chain-fixture/ledger/batch')
+        adapter.lock=threading.RLock();adapter.resource_lock=lambda *args:contextlib.nullcontext()
+        adapter.formal_done={};adapter.exit_registry={f'n{i}.example.invalid':dict(expected_exit=f'192.0.2.{i}',country='JP') for i in [1,2]}
+        source={'user_id':1,'owned':False,'token':'fixture-only','nodes':[dict(id=i,type='anytls',tag='node-'+str(i),relay='relay.example.invalid',relay_port=47000+i,sni=f'n{i}.example.invalid',uuid='fixture-only') for i in [1,2]]}
+        accounts=[]
+        def account_command(group,action,apply=False,client=None):
+            accounts.append(action)
+            return copy.deepcopy(source) if action in ['source','snapshot'] else {'state':'closed'}
+        adapter.account_command=account_command
+        now=int(time.time())
+        raw=json.dumps({'state':'live_verified','verified_at':now,'results':[dict(tag='node-'+str(i),relay='relay.example.invalid',relay_port=47000+i,sni=f'n{i}.example.invalid',exit_ip=f'192.0.2.{i}',country='JP',live=True,strict_tls=True) for i in [1,2]]}).encode()
+        files={};persisted=[];programs=[]
+        def execute(client,command,timeout=120):
+            if command.startswith("python3 - <<'DPLFORMAL'\n"):
+                program=command.split('\n',1)[1].rsplit('\nDPLFORMAL',1)[0]
+                compile(program,'actual-formal-remote','exec');programs.append(program)
+                files['/qa/formal-group-7/report.json']=raw
+            return ''
+        def php(client,body):
+            return {} if '$out=[]' in body else {'id':9,'token':'fixture-only','banned':True,'excluded':True}
+        def write(client,path,value):files[path]=value.encode() if isinstance(value,str) else value
+        adapter.atomic_json=lambda path,value:persisted.append(value)
+        class Channel:
+            def shutdown_write(self):pass
+            def recv_exit_status(self):return 0
+        class In:
+            channel=Channel()
+            def write(self,data):pass
+        class Out(io.BytesIO):channel=Channel()
+        class Gateway:
+            def exec_command(self,command,timeout=0):
+                if command!='python3 -':raise AssertionError('Unexpected QA probe command')
+                return In(),Out(b'403\n'),Out()
+        gateway=Gateway()
+        adapter.formal(candidate,None,gateway,php,execute,lambda client,path:files.get(path),write)
+        self.assertEqual(len(programs),1)
+        self.assertEqual(len(persisted),1)
+        self.assertEqual(len(persisted[0]['snapshot']),2,'The old node must remain in complete acceptance')
+        self.assertTrue(all(n['type']=='anytls' for n in persisted[0]['snapshot']))
+        self.assertEqual(persisted[0]['qas']['users']['new']['http_status'],403)
+        self.assertIn('cleanup',accounts)
+        self.assertEqual(adapter.formal(candidate,None,gateway,php,execute,lambda client,path:files.get(path),write)['nodes'],2)
+        self.assertEqual(len(programs),1,'A cached audit rechecks topology without rerunning the client')
+
     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)}