184 lines
14 KiB
Python
184 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Manual cfn subscription validator. No timers, notifications, or proxy traffic."""
|
|
import argparse, base64, collections, datetime, html, json, os, re, shlex, shutil, subprocess, sys, tempfile
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit, unquote
|
|
import yaml
|
|
|
|
HOSTS = {'article': '45.153.247.44', 'files': '45.142.166.119'}
|
|
FORMATS = {'clash': '.yaml', 'singbox': '.txt', 'v2ray': '.txt'}
|
|
MAX_BYTES = 20 * 1024 * 1024
|
|
|
|
def command(args, **kw):
|
|
return subprocess.run(args, capture_output=True, timeout=kw.pop('timeout', 45), **kw)
|
|
|
|
def ssh(host, code, env):
|
|
r = command(['ssh', '-o', 'StrictHostKeyChecking=yes', '-o', 'UserKnownHostsFile='+env['CFN_KNOWN_HOSTS'], '-o', 'ConnectTimeout=12', '-o', 'NumberOfPasswordPrompts=1', 'root@'+host, 'python3 -c '+shlex.quote(code)], env=env, timeout=60)
|
|
if r.returncode:
|
|
raise RuntimeError('source_ssh_failed:'+host+':exit_'+str(r.returncode))
|
|
if len(r.stdout) > MAX_BYTES: raise RuntimeError('source_response_too_large')
|
|
return r.stdout
|
|
|
|
def collect(root, secrets):
|
|
root.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix='auth-', dir=root) as auth:
|
|
a = Path(auth); (a/'credentials.json').write_text(json.dumps(secrets)); (a/'known_hosts').write_text(secrets['known_hosts'])
|
|
ask = a/'askpass.py'
|
|
ask.write_text('#!/usr/bin/python3\nimport os,sys,json\np=" ".join(sys.argv[1:])\nd=json.load(open(os.environ["CFN_AUTH_FILE"]))\nh=next((h for h in d["passwords"] if h in p),None)\nif not h:sys.exit(1)\nprint(d["passwords"][h])\n')
|
|
ask.chmod(0o700)
|
|
for f in [a/'credentials.json', a/'known_hosts']: f.chmod(0o600)
|
|
env = dict(os.environ, SSH_ASKPASS=str(ask), SSH_ASKPASS_REQUIRE='force', DISPLAY='1', CFN_AUTH_FILE=str(a/'credentials.json'), CFN_KNOWN_HOSTS=str(a/'known_hosts'))
|
|
php = r'''chdir('/www/wwwroot/clashfreenode-new.com');require 'vendor/autoload.php';Dotenv\Dotenv::createImmutable(getcwd())->load();try{$p=new PDO('mysql:host='.$_ENV['DB_HOST'].';port='.($_ENV['DB_PORT']??3306).';dbname='.$_ENV['DB_DATABASE'].';charset=utf8mb4',$_ENV['DB_USERNAME'],$_ENV['DB_PASSWORD'],[PDO::ATTR_ERRMODE=>PDO::ERRMODE_EXCEPTION]);$p->exec('SET TRANSACTION READ ONLY');$p->beginTransaction();$r=$p->query("SELECT a.id,a.sSlug,a.sContent,a.updated_at FROM articles a JOIN categories c ON c.id=a.iCategoryId WHERE c.sSlug='freenode' AND a.iStatus=1 ORDER BY a.id DESC LIMIT 1")->fetch(PDO::FETCH_ASSOC);echo json_encode($r);$p->rollBack();}catch(Throwable $e){echo json_encode(['error'=>'article_query_failed']);exit(1);}'''
|
|
code = 'import subprocess,sys\nr=subprocess.run(["php","-r",'+repr(php)+'],capture_output=True)\nsys.stdout.buffer.write(r.stdout)\nsys.exit(r.returncode)'
|
|
row = json.loads(ssh(HOSTS['article'], code, env)); content = row.pop('sContent')
|
|
links = list(dict.fromkeys(html.unescape(x) for x in re.findall(r'''(?:href|data-clipboard-text)=["']([^"']+)["']''', content)))
|
|
found = {}
|
|
for link in links:
|
|
u=urlsplit(link)
|
|
if u.scheme != 'https' or u.netloc != 'nodebus.net' or u.query or u.fragment: continue
|
|
m=re.fullmatch(r'/cfn/[a-f0-9]{16}/(\d{8})-(v2ray|singbox|clash)(\.txt|\.yaml)',u.path)
|
|
if not m or m[3] != FORMATS[m[2]]: continue
|
|
if m[2] in found and found[m[2]]['path'] != u.path: raise RuntimeError('ambiguous_subscription_links')
|
|
found[m[2]]={'path':u.path,'date':m[1]}
|
|
if set(found) != set(FORMATS): raise RuntimeError('missing_or_unrecognized_subscription_links')
|
|
article_date = re.fullmatch(r'fn(\d{8})', row['sSlug'])
|
|
if not article_date or any(v['date'] != article_date[1] for v in found.values()): raise RuntimeError('article_file_date_mismatch')
|
|
for kind,item in found.items():
|
|
code='from pathlib import Path\nimport sys\np=Path('+repr('/www/wwwroot/nodebus.net/public'+item['path'])+')\nr=p.resolve()\nassert r.is_relative_to(Path("/www/wwwroot/nodebus.net/public/cfn"))\nassert 0<p.stat().st_size<='+str(MAX_BYTES)+'\nsys.stdout.buffer.write(p.read_bytes())'
|
|
data=ssh(HOSTS['files'], code, env); f=root/(kind+FORMATS[kind]);f.write_bytes(data);f.chmod(0o600)
|
|
manifest={'article_id':row['id'],'article_slug':row['sSlug'],'article_updated_at_raw':row['updated_at'],'date':article_date[1], 'acquisition':'ssh-server-file', 'http_download':'not_tested_missing_registered_proxy', 'formats':list(found)}
|
|
(root/'manifest.json').write_text(json.dumps(manifest,ensure_ascii=False,indent=2));return manifest
|
|
|
|
def b64(s):
|
|
s=''.join(s.split());return base64.b64decode(s+'='*((-len(s))%4),validate=True)
|
|
|
|
def validate(kind, data):
|
|
issues=[]; result={'format':kind,'bytes':len(data),'issues':issues}; private=[]
|
|
def issue(code, **where): issues.append(dict(code=code,**where))
|
|
def remember(node):
|
|
for k in ['name','tag','server','password','uuid','id','username','sni','servername']:
|
|
v=node.get(k)
|
|
if isinstance(v,str) and v:private.append(v)
|
|
try:
|
|
s=data.decode('utf-8')
|
|
if not s.strip(): raise ValueError('empty')
|
|
if s.lstrip().lower().startswith(('<!doctype html','<html')): raise ValueError('html')
|
|
if kind=='clash':
|
|
d=yaml.safe_load(s)
|
|
if not isinstance(d,dict) or not isinstance(d.get('proxies'),list):raise ValueError('shape')
|
|
nodes=d['proxies']; result['nodes']=len(nodes);result['types']=dict(collections.Counter(str(n.get('type')) for n in nodes if isinstance(n,dict)))
|
|
names=[]
|
|
for i,n in enumerate(nodes):
|
|
if not isinstance(n,dict):issue('node_not_object',index=i);continue
|
|
remember(n);names.append(n.get('name'))
|
|
for k in ['name','type','server']:
|
|
if not isinstance(n.get(k),str) or not n[k].strip():issue('missing_or_empty_field',index=i,field=k)
|
|
check_port(n.get('port'),issue,i,'port')
|
|
required={'vmess':['uuid'],'vless':['uuid'],'trojan':['password'],'ss':['cipher','password'],'ssr':['cipher','password','protocol','obfs']}.get(n.get('type'),[])
|
|
for k in required:
|
|
if not n.get(k):issue('missing_or_empty_field',index=i,field=k)
|
|
dup=len(names)-len(set(str(x) for x in names))
|
|
if dup:issue('duplicate_node_names',count=dup)
|
|
groups=d.get('proxy-groups',[]);targets={str(n) for n in names}|{str(g.get('name')) for g in groups if isinstance(g,dict)}|{'DIRECT','REJECT','REJECT-DROP','PASS','COMPATIBLE','GLOBAL'}
|
|
for i,g in enumerate(groups):
|
|
for j,n in enumerate(g.get('proxies',[])):
|
|
if str(n) not in targets:issue('missing_group_reference',group_index=i,reference_index=j)
|
|
elif kind=='singbox':
|
|
d=json.loads(s)
|
|
if not isinstance(d,dict) or not isinstance(d.get('outbounds'),list):raise ValueError('shape')
|
|
nodes=d['outbounds'];result['outbounds']=len(nodes);result['nodes']=sum(n.get('type') not in ['selector','urltest','direct','block','dns'] for n in nodes if isinstance(n,dict));result['types']=dict(collections.Counter(str(n.get('type')) for n in nodes if isinstance(n,dict)))
|
|
tags=[]
|
|
for i,n in enumerate(nodes):
|
|
if not isinstance(n,dict):issue('node_not_object',index=i);continue
|
|
remember(n);tags.append(n.get('tag'))
|
|
if n.get('type') in ['selector','urltest','direct','block','dns']:continue
|
|
if not isinstance(n.get('server'),str) or not n['server'].strip():issue('missing_or_empty_field',index=i,field='server')
|
|
check_port(n.get('server_port'),issue,i,'server_port')
|
|
for k in {'vmess':['uuid'],'vless':['uuid'],'trojan':['password'],'shadowsocks':['method','password']}.get(n.get('type'),[]):
|
|
if not n.get(k):issue('missing_or_empty_field',index=i,field=k)
|
|
if len(tags)!=len(set(str(x) for x in tags)):issue('duplicate_outbound_tags')
|
|
else:
|
|
s=b64(s).decode('utf-8') if '://' not in s else s
|
|
lines=[x.strip() for x in s.splitlines() if x.strip()];result['nodes']=len(lines);schemes=collections.Counter()
|
|
for i,line in enumerate(lines,1):
|
|
if '\ufeff' in line:issue('unexpected_BOM',line=i)
|
|
line=line.lstrip('\ufeff');scheme,sep,payload=line.partition('://');schemes[scheme]+=1
|
|
if not sep or scheme not in ['vmess','vless','ss','ssr','trojan','hysteria','hysteria2','hy2','tuic','anytls','socks','socks5','http','https']:issue('unsupported_or_missing_scheme',line=i);continue
|
|
if not payload:issue('empty_payload',line=i);continue
|
|
if scheme=='vmess':
|
|
if unquote(payload).lstrip().startswith('{'):
|
|
issue('nonstandard_vmess_json_payload_requires_client_check',line=i)
|
|
continue
|
|
payload=payload.split('#',1)[0].split('?',1)[0]
|
|
try:
|
|
n=json.loads(b64(payload).decode());remember(n)
|
|
if not all(str(n.get(k,'')).strip() for k in ['add','id']):issue('vmess_required_field',line=i)
|
|
check_port(n.get('port'),issue,i,'port',location='line')
|
|
except Exception:issue('vmess_payload_not_supported_or_invalid',line=i)
|
|
elif scheme not in ['ss','ssr']:
|
|
try:
|
|
u=urlsplit(line)
|
|
if not u.hostname:issue('missing_uri_host',line=i)
|
|
check_port(u.port,issue,i,'port',location='line')
|
|
except Exception:issue('invalid_uri_authority',line=i)
|
|
result['schemes']=dict(schemes);result['core_check']={'status':'not_applicable','reason':'URI list requires client importer; no whole-file core check claimed'}
|
|
if not result.get('nodes'):issue('empty_node_list')
|
|
result['parse']='pass'
|
|
except Exception as e:
|
|
result['parse']='fail';issue('parse_error',error_type=type(e).__name__)
|
|
if hasattr(e,'lineno'):issues[-1]['line']=e.lineno
|
|
if hasattr(e,'problem_mark') and e.problem_mark:issues[-1]['line']=e.problem_mark.line+1
|
|
result['structure_status']='fail' if issues else 'pass';result['issue_count']=len(issues);return result,private
|
|
|
|
def check_port(v, issue, index, field, location='index'):
|
|
try:ok=not isinstance(v,bool) and str(v).isdigit() and 1<=int(v)<=65535
|
|
except Exception:ok=False
|
|
if not ok:issue('invalid_port',**{location:index,'field':field})
|
|
|
|
def core_check(kind, path, root, binaries, private):
|
|
core=binaries/('mihomo' if kind=='clash' else 'sing-box')
|
|
if not core.exists():return {'status':'not_tested','reason':'core_missing'}
|
|
home=root/('core-'+kind);home.mkdir(mode=0o700,exist_ok=True)
|
|
if kind=='clash':
|
|
for name in ['Country.mmdb','geoip.metadb']:
|
|
if (binaries/name).exists():shutil.copyfile(binaries/name,home/name)
|
|
args=[str(core),'-t','-d',str(home),'-f',str(path)] if kind=='clash' else [str(core),'check','-D',str(home),'-c',str(path)]
|
|
# A private network blocks rule/provider downloads and all node traffic. No proxy is started.
|
|
wrapper=['systemd-run','--quiet','--wait','--pipe','--collect','--service-type=exec','-p','MemoryMax=256M','-p','MemorySwapMax=0','-p','CPUQuota=50%','-p','RuntimeMaxSec=45','-p','PrivateNetwork=yes','-p','NoNewPrivileges=yes','-p','ProtectSystem=strict','-p','ReadWritePaths='+str(home),'-p','TasksMax=64','/usr/bin/time','-f','CHECKER_MAX_RSS_KB=%M']
|
|
try:
|
|
r=command(wrapper+args,timeout=60);text=(r.stdout+r.stderr).decode(errors='replace')
|
|
mem=re.search(r'CHECKER_MAX_RSS_KB=(\d+)',text)
|
|
for v in sorted(set(private),key=len,reverse=True):
|
|
if len(v)>2:text=text.replace(v,'[redacted]')
|
|
text=re.sub(r'(?:https?|ss|vmess|vless|trojan)://[^\s"\']+','[redacted-url]',text)
|
|
text=re.sub(r'\x1b\[[0-9;]*m','',text)
|
|
status='pass' if r.returncode==0 else 'fail'
|
|
if r.returncode and any(x in text.lower() for x in ['network is unreachable','network is down','no such host','download','geosite.dat','geoip.dat','country.mmdb']):status='blocked_dependency'
|
|
return {'status':status,'exit_code':r.returncode,'peak_rss_kb':int(mem[1]) if mem else None,'message':text[-3500:]}
|
|
except subprocess.TimeoutExpired:return {'status':'timeout'}
|
|
|
|
def main():
|
|
os.umask(0o077)
|
|
a=argparse.ArgumentParser();a.add_argument('--state-dir',default='/var/lib/cfn-subscription-checker');a.add_argument('--binaries',default='/opt/cfn-checker-cores');a.add_argument('--snapshot-dir');a.add_argument('--credentials-stdin',action='store_true');args=a.parse_args()
|
|
now=datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8)));root=Path(args.state_dir)/now.strftime('%Y%m%d-%H%M%S-%f');root.mkdir(mode=0o700,parents=True)
|
|
report={'checked_at':now.isoformat(),'http_download':'not_tested_missing_registered_proxy','node_connectivity':'not_tested','results':[]}
|
|
try:
|
|
if args.snapshot_dir:
|
|
source=Path(args.snapshot_dir);manifest=json.loads((source/'manifest.json').read_text())
|
|
for k,suffix in FORMATS.items():shutil.copyfile(source/(k+suffix),root/(k+suffix))
|
|
elif args.credentials_stdin:manifest=collect(root,json.load(sys.stdin))
|
|
else:raise RuntimeError('credentials_stdin_or_snapshot_dir_required')
|
|
report['source']=manifest;report['freshness']='pass' if manifest['date']==now.strftime('%Y%m%d') else 'stale'
|
|
for kind,suffix in FORMATS.items():
|
|
f=root/(kind+suffix);f.chmod(0o600);result,private=validate(kind,f.read_bytes())
|
|
if kind!='v2ray':result['core_check']=core_check(kind,f,root,Path(args.binaries),private)
|
|
report['results'].append(result)
|
|
failed=report['freshness']!='pass' or any(r['structure_status']=='fail' or r['core_check']['status'] in ['fail','timeout'] for r in report['results'])
|
|
incomplete=any(r['core_check']['status'] in ['blocked_dependency','not_tested'] for r in report['results'])
|
|
report['status']='fail' if failed else 'incomplete' if incomplete else 'pass_server_files_only'
|
|
except Exception as e:
|
|
report['status']='acquisition_failed';report['error_type']=type(e).__name__;report['error']=str(e) if isinstance(e,RuntimeError) else 'see source or format stage'
|
|
output=root/'report.json';output.write_text(json.dumps(report,ensure_ascii=False,indent=2));print(output);print(json.dumps(report,ensure_ascii=False,indent=2));return 0 if report['status']=='pass_server_files_only' else 1
|
|
|
|
if __name__=='__main__':sys.exit(main())
|