Files
admin e6207aa691
build / build-and-scan (push) Waiting to run
chore(仓库对齐): 文档库结构治理 + IM/插件线落地
文档库:目录改为编号制(01-规范/02-架构设计/03-数据库/04-调整方案/
05-交接单/06-ops/07-scripts/08-skills/09-archive),顶层散文件归入 01-规范/;
INDEX.md 与 docs-manifest.json 重刷(档案 146 篇);旧目录名引用全量对齐。

IM 线:src/im/**(SDK / hub / store / presence / ws / gateway-token)、
src/web/routes/im.ts、src/db/plugin-data/**、src/supervisor/plugin-assembly.ts
及对应 test/**。

插件线:poc/{im-agent-bridge,im-connection-gateway,im-conversation-tabs,
business-plugins-im,carbon-mcp-probe}、src/web/routes/{sessions,overlay-device}.ts、
src/net/relay/{device-grant,instance-credential}.ts。

仓库卫生:清出 40 个历史误入库 / 已改名文件(34 个交接单归档 + 6 个旧结构,
本地均有副本);dsh-server-docs/.gitignore 补 tmp/;交接单不入库(政策)。
2026-09-24 07:25:16 +08:00

89 lines
3.2 KiB
Python

#!/usr/bin/env python3
"""探测 _acme-challenge 名称是否被通配 CNAME 劫持(只读+临时记录,用完即删)。
用法: python3 cf-probe.py <zone名> [_acme-challenge.<zone>]
"""
import json
import re
import sys
import time
import urllib.request
CRED = '/etc/cloudflare.ini'
# 2026-09-19 域迁 ai1net.com 后其 zone 走独立凭据(与旧域 token 不通用)
CREDS = {'ai1net.com': '/etc/cloudflare-ai1net.ini'}
API = 'https://api.cloudflare.com/client/v4'
def token(zone=None):
txt = open(CREDS.get(zone or '', CRED), 'r', encoding='utf-8').read()
m = re.search(r'dns_cloudflare_api_token\s*=\s*([A-Za-z0-9_\-]+)', txt)
if not m:
sys.exit('未找到 token')
return m.group(1)
def req(method, path, tok, body=None):
data = json.dumps(body).encode('utf-8') if body is not None else None
r = urllib.request.Request(API + path, data=data, method=method,
headers={'Authorization': 'Bearer ' + tok,
'Content-Type': 'application/json'})
with urllib.request.urlopen(r, timeout=20) as resp:
return json.loads(resp.read().decode('utf-8'))
def doh(name, typ, server='https://dns.google/resolve'):
url = '%s?name=%s&type=%s' % (server, name, typ)
r = urllib.request.Request(url, headers={'accept': 'application/dns-json'})
with urllib.request.urlopen(r, timeout=20) as resp:
return json.loads(resp.read().decode('utf-8'))
def show(title, name, typ):
print('--- %s ---' % title)
for srv, label in [('https://dns.google/resolve', 'Google'), ('https://cloudflare-dns.com/dns-query', 'CF')]:
try:
d = doh(name, typ, srv)
ans = d.get('Answer') or []
print(' [%s] Status=%s' % (label, d.get('Status')))
if not ans:
print(' (无 Answer)')
for a in ans:
print(' type=%-6s %s' % (a.get('type'), str(a.get('data'))[:90]))
except Exception as e:
print(' [%s] 查询失败: %s' % (label, e))
def main():
zone = sys.argv[1] if len(sys.argv) > 1 else 'ai1net.com'
name = sys.argv[2] if len(sys.argv) > 2 else '_acme-challenge.' + zone
tok = token(zone)
zid = ''
for z in (req('GET', '/zones?name=' + zone, tok).get('result') or []):
zid = z['id']
if zid == '':
sys.exit('zone 不在权限内')
print('=== 创建临时 TXT: %s ===' % name)
created = req('POST', '/zones/%s/dns_records' % zid, tok,
{'type': 'TXT', 'name': name, 'content': 'probe-test-value-12345', 'ttl': 120})
if not created.get('success'):
print(' 创建失败:', created.get('errors'))
return
rid = created['result']['id']
print(' 已创建 id=%s' % rid)
try:
for wait in (3, 10, 30):
time.sleep(wait if wait == 3 else wait - 3)
print('\n=== 等待累计 ~%ds 后查询 ===' % wait)
show('TXT 查询', name, 'TXT')
finally:
print('\n=== 清理临时记录 ===')
d = req('DELETE', '/zones/%s/dns_records/%s' % (zid, rid), tok)
print(' 删除成功:', d.get('success'))
if __name__ == '__main__':
main()