Add reviewed append-only node operations and batch coordination
This commit is contained in:
hellcat
parent
b05c321550
commit
b8d10fac96
368
batch.py
Normal file
368
batch.py
Normal file
@ -0,0 +1,368 @@
|
||||
#!/usr/bin/env python3
|
||||
"""DPL append-only batch coordinator. Adapters must probe before and after writes."""
|
||||
import argparse
|
||||
import concurrent.futures
|
||||
import contextlib
|
||||
import fcntl
|
||||
import hashlib
|
||||
import ipaddress
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import signal
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
PHASES = ('stage', 'dns', 'provision', 'relay', 'verify', 'publish', 'audit')
|
||||
SHARED = {'stage': 'panel', 'dns': 'dns', 'relay': 'relay', 'publish': 'panel'}
|
||||
STOP = threading.Event()
|
||||
|
||||
|
||||
def digest(value):
|
||||
return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False,
|
||||
separators=(',', ':')).encode()).hexdigest()
|
||||
|
||||
|
||||
def read_json(path):
|
||||
return json.loads(Path(path).read_text())
|
||||
|
||||
|
||||
def atomic_json(path, value):
|
||||
path = Path(path)
|
||||
tmp = path.with_name(path.name + '.tmp-' + str(os.getpid()) + '-' + str(threading.get_ident()))
|
||||
with tmp.open('x') as f:
|
||||
os.chmod(tmp, 0o600)
|
||||
json.dump(value, f, ensure_ascii=False, sort_keys=True)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.replace(tmp, path)
|
||||
|
||||
|
||||
def validate(plan):
|
||||
if plan.get('version') != 1 or plan.get('mode') != 'append':
|
||||
raise ValueError('Only version=1, mode=append is supported')
|
||||
if not re.fullmatch(r'[a-z0-9-]{1,64}', plan.get('batch_id', '')):
|
||||
raise ValueError('Invalid batch_id')
|
||||
context = plan.get('context', {})
|
||||
required = ('panel_vps', 'gateway_vps', 'jump_vps', 'app_root', 'panel_url',
|
||||
'subscription_url', 'provision_commit', 'bundle', 'client_binary')
|
||||
if any(not context.get(k) for k in required):
|
||||
raise ValueError('Missing explicit context fields')
|
||||
if not re.fullmatch(r'[0-9a-f]{40}', context['provision_commit']):
|
||||
raise ValueError('A full reviewed, pushed Git commit is required')
|
||||
for k in ('panel_url', 'subscription_url'):
|
||||
if not re.fullmatch(r'https://[a-z0-9.-]+', context[k]):
|
||||
raise ValueError('HTTPS origin without token/path required: ' + k)
|
||||
adapter = plan.get('adapter')
|
||||
if not isinstance(adapter, list) or not adapter or not all(isinstance(x, str) for x in adapter):
|
||||
raise ValueError('adapter must be an argv array, never a shell string')
|
||||
if not plan.get('nodes') or not isinstance(plan['nodes'], list):
|
||||
raise ValueError('A nonempty node list is required')
|
||||
seen = {k: set() for k in ('key', 'name', 'vps_id', 'node_address', 'tls_domain', 'relay', 'socks_port', 'number')}
|
||||
for n in plan['nodes']:
|
||||
if any(k not in n for k in ('key', 'country', 'sequence', 'flag', 'country_name', 'name',
|
||||
'vps_id', 'node_address', 'listen_port', 'relay_host',
|
||||
'relay_port', 'tls_domain', 'formal_group_id',
|
||||
'expected_exit', 'socks_port')):
|
||||
raise ValueError('Incomplete node specification')
|
||||
if not re.fullmatch(r'[a-z0-9-]{1,48}', n['key']):
|
||||
raise ValueError('Invalid deployment key')
|
||||
if not re.fullmatch(r'[A-Z]{2}', n['country']):
|
||||
raise ValueError('Country must be a two-letter code')
|
||||
if type(n['sequence']) is not int or not 1 <= n['sequence'] <= 99:
|
||||
raise ValueError('Display sequence must be 01..99')
|
||||
expected = f"{n['flag']} {n['country_name']} {n['country']} - {n['sequence']:02d}"
|
||||
flag = ''.join(chr(0x1F1E6 + ord(x) - ord('A')) for x in n['country'])
|
||||
if n['name'] != expected or n['flag'] != flag:
|
||||
raise ValueError('Display name/flag does not match country and sequence')
|
||||
for k in ('listen_port', 'relay_port', 'socks_port'):
|
||||
if type(n[k]) is not int or not 1024 <= n[k] <= 65535:
|
||||
raise ValueError('Invalid port: ' + k)
|
||||
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'])
|
||||
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)
|
||||
for k in ('vps_id',):
|
||||
if not re.fullmatch(r'VPS-[0-9]{3}', n[k]):
|
||||
raise ValueError('Use the exact CoDocs VPS ID')
|
||||
identities = {k: n[k] for k in ('key', 'name', 'vps_id', 'node_address', 'tls_domain', 'socks_port')}
|
||||
identities.update(relay=(n['relay_host'], n['relay_port']), number=(n['country'], n['sequence']))
|
||||
for k, value in identities.items():
|
||||
if value in seen[k]:
|
||||
raise ValueError('Duplicate batch identity: ' + k)
|
||||
seen[k].add(value)
|
||||
# The plan carries references, never secret values. Adapters load credentials themselves.
|
||||
def check(value):
|
||||
if isinstance(value, dict):
|
||||
for k, v in value.items():
|
||||
if re.search(r'password|secret|token|private_key', k, re.I):
|
||||
raise ValueError('Secret fields are not allowed in a batch plan')
|
||||
check(v)
|
||||
elif isinstance(value, list):
|
||||
for v in value:
|
||||
check(v)
|
||||
check(plan)
|
||||
return plan
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def lock(root, name):
|
||||
path = root / ('lock-' + hashlib.sha256(name.encode()).hexdigest())
|
||||
with path.open('a+') as f:
|
||||
os.chmod(path, 0o600)
|
||||
while True:
|
||||
try:
|
||||
fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
break
|
||||
except BlockingIOError:
|
||||
if STOP.wait(.1):
|
||||
raise RuntimeError('Stopped while waiting for resource lock')
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
fcntl.flock(f, fcntl.LOCK_UN)
|
||||
|
||||
|
||||
class Ledger:
|
||||
def __init__(self, root):
|
||||
self.root = Path(root).resolve()
|
||||
if self.root == Path('/'):
|
||||
raise ValueError('Invalid state root')
|
||||
self.root.mkdir(parents=True, mode=0o700, exist_ok=True)
|
||||
os.chmod(self.root, 0o700)
|
||||
self.path = self.root / 'registry.sqlite3'
|
||||
with self.db() as c:
|
||||
c.executescript('''
|
||||
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 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)
|
||||
|
||||
def db(self):
|
||||
c = sqlite3.connect(self.path, timeout=30)
|
||||
c.execute('PRAGMA synchronous=FULL')
|
||||
return c
|
||||
|
||||
def reserve(self, plan):
|
||||
namespace = plan['context']['panel_vps'] + ':' + plan['context']['app_root']
|
||||
with self.db() as c:
|
||||
c.execute('BEGIN IMMEDIATE')
|
||||
prior = c.execute('SELECT digest FROM batches WHERE id=?', (plan['batch_id'],)).fetchone()
|
||||
if prior and prior[0] != digest(plan):
|
||||
raise ValueError('Batch plan changed; resume requires the original exact plan')
|
||||
for n in plan['nodes']:
|
||||
owner = digest({'context': plan['context'], 'node': n})
|
||||
claims = {k: n[k] for k in ('key', 'name', 'vps_id', 'node_address', 'tls_domain')}
|
||||
claims['relay'] = [n['relay_host'], n['relay_port']]
|
||||
claims['number'] = [n['country'], n['sequence']]
|
||||
for kind, value in claims.items():
|
||||
scope = namespace
|
||||
if kind == 'relay':
|
||||
scope = 'relay:' + n['relay_host']
|
||||
elif kind in ('vps_id', 'node_address', 'tls_domain'):
|
||||
scope = 'infrastructure'
|
||||
encoded = json.dumps(value, ensure_ascii=False, sort_keys=True)
|
||||
existing = c.execute('SELECT owner FROM claims WHERE namespace=? AND kind=? AND value=?',
|
||||
(scope, kind, encoded)).fetchone()
|
||||
if existing and existing[0] != owner:
|
||||
raise ValueError('Another batch reserved identity: ' + kind)
|
||||
c.execute('INSERT OR IGNORE INTO claims VALUES(?,?,?,?)', (scope, kind, encoded, owner))
|
||||
# SOCKS listeners are shared by all batches using this gateway.
|
||||
scope, value = 'gateway:' + plan['context']['gateway_vps'], str(n['socks_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('Another batch reserved the gateway SOCKS port')
|
||||
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):
|
||||
with self.db() as c:
|
||||
row = c.execute('SELECT status FROM phases WHERE batch=? AND node=? AND phase=?',
|
||||
(batch, key, phase)).fetchone()
|
||||
return row[0] if row else 'pending'
|
||||
|
||||
def mark(self, batch, key, phase, status, attempt=False):
|
||||
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)))
|
||||
|
||||
|
||||
class Coordinator:
|
||||
def __init__(self, plan, root, jobs=4, timeout=900):
|
||||
self.plan = validate(plan)
|
||||
self.ledger = Ledger(root)
|
||||
self.jobs = jobs
|
||||
self.timeout = timeout
|
||||
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)
|
||||
|
||||
def call(self, n, phase, action):
|
||||
fingerprint = digest({'context': self.plan['context'], 'node': n})
|
||||
path = self.dir / (n['key'] + '.json')
|
||||
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)]
|
||||
with subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
|
||||
start_new_session=True) as p:
|
||||
deadline = time.monotonic() + self.timeout
|
||||
while True:
|
||||
try:
|
||||
output, _ = p.communicate(timeout=.2)
|
||||
break
|
||||
except subprocess.TimeoutExpired:
|
||||
if STOP.is_set() or time.monotonic() >= deadline:
|
||||
os.killpg(p.pid, signal.SIGTERM)
|
||||
try:
|
||||
p.communicate(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
os.killpg(p.pid, signal.SIGKILL)
|
||||
p.communicate()
|
||||
raise RuntimeError('Adapter interrupted; remote outcome must be probed')
|
||||
if p.returncode or len(output) > 65536:
|
||||
raise RuntimeError('Adapter failed; no stdout/stderr persisted')
|
||||
r = json.loads(output)
|
||||
if (r.get('key'), r.get('phase'), r.get('request_sha256')) != (n['key'], phase, fingerprint):
|
||||
raise RuntimeError('Adapter receipt identity mismatch')
|
||||
if r.get('status') not in ('ready', 'absent', 'waiting', 'unknown', 'conflict'):
|
||||
raise RuntimeError('Invalid adapter status')
|
||||
return r
|
||||
|
||||
def phase(self, n, phase):
|
||||
try:
|
||||
return self._phase(n, phase)
|
||||
except Exception:
|
||||
self.ledger.mark(self.plan['batch_id'], n['key'], phase, 'unknown')
|
||||
return False
|
||||
|
||||
def _phase(self, n, phase):
|
||||
batch, key = self.plan['batch_id'], n['key']
|
||||
resource = SHARED.get(phase, n['vps_id'])
|
||||
# Common panel/dns/relay lock names across batches and processes.
|
||||
namespace = self.plan['context']['panel_vps']
|
||||
if resource in ('dns', 'relay'):
|
||||
namespace = 'dpl-external'
|
||||
with lock(self.ledger.root, namespace + ':' + resource):
|
||||
r = self.call(n, phase, 'probe')
|
||||
if r['status'] == 'ready':
|
||||
self.ledger.mark(batch, key, phase, 'done')
|
||||
return True
|
||||
if r['status'] == 'conflict':
|
||||
STOP.set()
|
||||
if phase in ('preflight', 'verify', 'audit') or r['status'] != 'absent' or r.get('retry_safe') is not True:
|
||||
self.ledger.mark(batch, key, phase, r['status'])
|
||||
return False
|
||||
if STOP.is_set():
|
||||
return False
|
||||
# Persist intent before a write; never infer success from the command exit code.
|
||||
self.ledger.mark(batch, key, phase, 'running', attempt=True)
|
||||
applied = self.call(n, phase, 'apply')
|
||||
if applied['status'] == 'conflict':
|
||||
self.ledger.mark(batch, key, phase, 'conflict')
|
||||
STOP.set()
|
||||
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'] == 'conflict':
|
||||
STOP.set()
|
||||
return r['status'] == 'ready'
|
||||
|
||||
def worker(self, n):
|
||||
try:
|
||||
with lock(self.ledger.root, 'node:' + self.plan['context']['panel_vps'] + ':' + n['key']):
|
||||
self.ledger.mark(self.plan['batch_id'], n['key'], 'coordinator', 'running')
|
||||
# Already finished batches are re-audited; disabled QA need not be re-enabled.
|
||||
if self.ledger.state(self.plan['batch_id'], n['key'], 'audit') == 'done':
|
||||
success = self.phase(n, 'audit')
|
||||
else:
|
||||
success = True
|
||||
for phase in PHASES:
|
||||
if STOP.is_set() or not self.phase(n, phase):
|
||||
success = False
|
||||
break
|
||||
self.ledger.mark(self.plan['batch_id'], n['key'], 'coordinator', 'done' if success else 'waiting')
|
||||
return success
|
||||
except Exception:
|
||||
# Never print adapter payloads, commands, credentials or exception contents.
|
||||
self.ledger.mark(self.plan['batch_id'], n['key'], 'coordinator', 'unknown')
|
||||
return False
|
||||
|
||||
def run(self):
|
||||
# Whole-batch barrier: no staging until every candidate passed current preflight.
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=self.jobs) as pool:
|
||||
def check(n):
|
||||
try:
|
||||
return self.phase(n, 'preflight')
|
||||
except Exception:
|
||||
self.ledger.mark(self.plan['batch_id'], n['key'], 'preflight', 'unknown')
|
||||
return False
|
||||
checks = list(pool.map(check, self.plan['nodes']))
|
||||
if not all(checks):
|
||||
return False
|
||||
return all(list(pool.map(self.worker, self.plan['nodes'])))
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser(description=__doc__)
|
||||
p.add_argument('command', choices=('plan', 'run', 'status'))
|
||||
p.add_argument('--plan', required=True)
|
||||
p.add_argument('--state-root', required=True)
|
||||
p.add_argument('--jobs', type=int, default=4)
|
||||
p.add_argument('--timeout', type=int, default=900)
|
||||
p.add_argument('--execute', action='store_true', help='Required for a run that may write remotely')
|
||||
a = p.parse_args()
|
||||
os.umask(0o077)
|
||||
if not 1 <= a.jobs <= 16 or not 1 <= a.timeout <= 3600:
|
||||
p.error('jobs must be 1..16; timeout 1..3600')
|
||||
try:
|
||||
plan = validate(read_json(a.plan))
|
||||
if a.command == 'plan':
|
||||
print(json.dumps({'batch_id': plan['batch_id'], 'nodes': len(plan['nodes']),
|
||||
'plan_sha256': digest(plan), 'writes': False}))
|
||||
return 0
|
||||
if a.command == 'run' and not a.execute:
|
||||
p.error('run requires --execute; plan does not perform remote requests')
|
||||
if a.command == 'status':
|
||||
path = Path(a.state_root).resolve() / 'registry.sqlite3'
|
||||
rows = {}
|
||||
if path.exists():
|
||||
with sqlite3.connect(path.as_uri() + '?mode=ro', uri=True) as c:
|
||||
prior = c.execute('SELECT digest FROM batches WHERE id=?', (plan['batch_id'],)).fetchone()
|
||||
if prior and prior[0] != digest(plan):
|
||||
raise ValueError('Batch plan differs from its reserved plan')
|
||||
rows = {(key, phase): status for key, phase, status in c.execute(
|
||||
'SELECT node,phase,status FROM phases WHERE batch=?', (plan['batch_id'],))}
|
||||
print(json.dumps({n['key']: {phase: rows.get((n['key'], phase), 'pending')
|
||||
for phase in ('preflight',) + PHASES + ('coordinator',)}
|
||||
for n in plan['nodes']}))
|
||||
return 0
|
||||
runner = Coordinator(plan, a.state_root, a.jobs, a.timeout)
|
||||
for sig in (signal.SIGINT, signal.SIGTERM):
|
||||
signal.signal(sig, lambda *_: STOP.set())
|
||||
success = runner.run()
|
||||
print(json.dumps({'batch_id': plan['batch_id'], 'complete': success,
|
||||
'next': None if success else 'Inspect status, resolve waiting/conflict, rerun the exact plan'}))
|
||||
return 0 if success else 2
|
||||
except Exception as exc:
|
||||
# Validation errors have no adapter/credential material.
|
||||
print(json.dumps({'error': str(exc) if isinstance(exc, ValueError) else 'Coordinator failed; inspect protected state'}))
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
66
command-adapter.py
Normal file
66
command-adapter.py
Normal file
@ -0,0 +1,66 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Run reviewed, pinned site-specific commands for batch.py without a shell.
|
||||
|
||||
Missing integration is a waiting gate, never an assumed successful operation.
|
||||
Handlers receive paths to the frozen request, not passwords on the command line.
|
||||
"""
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser(description=__doc__)
|
||||
for field in ('plan', 'node', 'phase', 'action', 'state-dir'):
|
||||
p.add_argument('--' + field, required=True)
|
||||
a = p.parse_args()
|
||||
node = json.loads(Path(a.node).read_text())
|
||||
plan = json.loads(Path(a.plan).read_text())
|
||||
identity = {'key': node['key'], 'phase': a.phase, 'request_sha256': node['request_sha256']}
|
||||
try:
|
||||
profile = plan['command_profile']
|
||||
# Every runnable local script/config must be pinned; pinning the interpreter
|
||||
# alone does not pin scripts, remote code or the commands' dependencies.
|
||||
files = profile['files']
|
||||
if not files:
|
||||
raise ValueError('No pinned handler files')
|
||||
for filename, expected in files.items():
|
||||
if not Path(filename).is_absolute():
|
||||
raise ValueError('Absolute handler paths required')
|
||||
if hashlib.sha256(Path(filename).read_bytes()).hexdigest() != expected:
|
||||
raise ValueError('Handler changed')
|
||||
command = profile['phases'].get(a.phase, {}).get(a.action)
|
||||
if command is None:
|
||||
result = dict(identity, status='waiting', reason='handler_not_configured')
|
||||
else:
|
||||
if not isinstance(command, list) or not command or not all(isinstance(x, str) for x in command):
|
||||
raise ValueError('Command must be argv')
|
||||
values = dict(plan=a.plan, node=a.node, state_dir=a.state_dir,
|
||||
key=node['key'], phase=a.phase, action=a.action)
|
||||
argv = [x.format_map(values) for x in command]
|
||||
if not Path(argv[0]).is_absolute():
|
||||
raise ValueError('Absolute executable required')
|
||||
completed = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
|
||||
if completed.returncode or len(completed.stdout) > 65536:
|
||||
raise ValueError('Handler failed')
|
||||
result = json.loads(completed.stdout)
|
||||
if any(result.get(k) != v for k, v in identity.items()):
|
||||
raise ValueError('Wrong handler receipt')
|
||||
if result.get('status') not in ('ready', 'absent', 'waiting', 'unknown', 'conflict'):
|
||||
raise ValueError('Wrong handler status')
|
||||
# Only structured nonsecret receipt fields are allowed through to the runner.
|
||||
safe = {**identity, 'status': result['status']}
|
||||
if result.get('retry_safe') is True:
|
||||
safe['retry_safe'] = True
|
||||
print(json.dumps(safe))
|
||||
return 0
|
||||
except Exception:
|
||||
print(json.dumps(dict(identity, status='unknown')))
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
332
operations.php
Normal file
332
operations.php
Normal file
@ -0,0 +1,332 @@
|
||||
<?php
|
||||
/** Shared CLI helpers; loading this file does not bootstrap Laravel or open a connection. */
|
||||
namespace DplOnboarding;
|
||||
|
||||
use App\Models\{Server, ServerGroup, User, Plan, Setting};
|
||||
use Illuminate\Support\Facades\{DB, Cache};
|
||||
use RuntimeException;
|
||||
|
||||
function ids(string $value): array
|
||||
{
|
||||
if (!preg_match('/^[1-9][0-9]*(,[1-9][0-9]*)*$/D', $value)) {
|
||||
throw new RuntimeException('Expected explicit comma-separated positive IDs');
|
||||
}
|
||||
$ids = array_map('intval', explode(',', $value));
|
||||
if (count($ids) !== count(array_unique($ids))) throw new RuntimeException('Duplicate IDs');
|
||||
return $ids;
|
||||
}
|
||||
|
||||
function readJson(string $path): array
|
||||
{
|
||||
if (!is_file($path) || is_link($path)) throw new RuntimeException('Required regular journal/manifest missing');
|
||||
$data = json_decode(file_get_contents($path), true, flags: JSON_THROW_ON_ERROR);
|
||||
if (!is_array($data)) throw new RuntimeException('Invalid JSON document');
|
||||
return $data;
|
||||
}
|
||||
|
||||
function writeJson(string $path, array $data): void
|
||||
{
|
||||
if (is_link($path)) throw new RuntimeException('Refusing journal symlink');
|
||||
$tmp = tempnam(dirname($path), '.onboarding-');
|
||||
if ($tmp === false) throw new RuntimeException('Cannot create journal');
|
||||
try {
|
||||
chmod($tmp, 0600);
|
||||
$body = json_encode($data, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE) . "\n";
|
||||
if (file_put_contents($tmp, $body) !== strlen($body) || !rename($tmp, $path)) {
|
||||
throw new RuntimeException('Cannot persist journal');
|
||||
}
|
||||
} finally {
|
||||
if (is_file($tmp)) unlink($tmp);
|
||||
}
|
||||
}
|
||||
|
||||
function manifest(string $dir): array
|
||||
{
|
||||
$m = readJson($dir . '/manifest.json');
|
||||
foreach (['node_id','machine_id','group_id','test_user_id'] as $field) {
|
||||
if (!isset($m[$field]) || !preg_match('/^[1-9][0-9]*$/D', (string)$m[$field])) {
|
||||
throw new RuntimeException('Manifest missing positive object ID');
|
||||
}
|
||||
$m[$field] = (int)$m[$field];
|
||||
}
|
||||
if (!isset($m['key']) || !preg_match('/^[a-z0-9-]+$/D', $m['key'])) throw new RuntimeException('Invalid manifest key');
|
||||
return $m;
|
||||
}
|
||||
|
||||
function locked(string $dir, callable $fn): mixed
|
||||
{
|
||||
if (!is_dir($dir) || is_link($dir) || (fileperms($dir)&0077)!==0) throw new RuntimeException('Existing private backup directory required');
|
||||
$path = $dir . '/operation.lock';
|
||||
if (is_link($path)) throw new RuntimeException('Refusing lock symlink');
|
||||
$h = fopen($path, 'c');
|
||||
if ($h === false) throw new RuntimeException('Cannot open operation lock');
|
||||
chmod($path, 0600);
|
||||
try {
|
||||
if (!flock($h, LOCK_EX | LOCK_NB)) throw new RuntimeException('Another operation is active');
|
||||
return $fn();
|
||||
} finally { flock($h, LOCK_UN); fclose($h); }
|
||||
}
|
||||
|
||||
function boot(): void
|
||||
{
|
||||
require getcwd() . '/vendor/autoload.php';
|
||||
$app = require getcwd() . '/bootstrap/app.php';
|
||||
$app->make(\Illuminate\Contracts\Console\Kernel::class)->bootstrap();
|
||||
}
|
||||
|
||||
final class PanelChange
|
||||
{
|
||||
private const MODELS = ['node'=>Server::class,'qa'=>User::class,'user'=>User::class,
|
||||
'plan'=>Plan::class,'group'=>ServerGroup::class,'setting'=>Setting::class];
|
||||
private const FIELDS = ['node'=>['group_ids'],'qa'=>['banned'],'user'=>['group_id'],
|
||||
'plan'=>['group_id'],'group'=>['name'],'setting'=>['value']];
|
||||
|
||||
private static function normalize(string $field, mixed $value): mixed
|
||||
{
|
||||
return match ($field) {
|
||||
'group_ids' => array_map('strval', $value ?? []),
|
||||
'group_id' => $value === null ? null : (int)$value,
|
||||
'banned','enabled','show' => (bool)$value,
|
||||
default => $value,
|
||||
};
|
||||
}
|
||||
|
||||
private static function object(string $kind, int|string $id): mixed
|
||||
{
|
||||
$class = self::MODELS[$kind] ?? throw new RuntimeException('Unsupported journal object');
|
||||
return $class::where($kind === 'setting' ? 'name' : 'id', $id)->lockForUpdate()->first();
|
||||
}
|
||||
|
||||
private static function value(mixed $model, string $field): mixed
|
||||
{
|
||||
return $model === null ? null : self::normalize($field, $model->getAttribute($field));
|
||||
}
|
||||
|
||||
private static function operation(string $kind, mixed $model, string $field, mixed $after, int|string|null $id = null): ?array
|
||||
{
|
||||
if (!in_array($field, self::FIELDS[$kind] ?? [], true)) throw new RuntimeException('Unsupported journal field');
|
||||
$before = self::value($model, $field);
|
||||
$after = self::normalize($field, $after);
|
||||
if ($model !== null && $before === $after) return null;
|
||||
return ['kind'=>$kind,'id'=>$id ?? $model->getKey(),'field'=>$field,
|
||||
'before_exists'=>$model !== null,'before'=>$before,'after'=>$after];
|
||||
}
|
||||
|
||||
private static function context(array $m): array
|
||||
{
|
||||
$node = self::object('node', $m['node_id']) ?? throw new RuntimeException('Manifest node missing');
|
||||
$qa = self::object('qa', $m['test_user_id']) ?? throw new RuntimeException('Manifest QA missing');
|
||||
if ((int)$node->machine_id !== $m['machine_id'] || $node->type !== 'anytls') {
|
||||
throw new RuntimeException('Manifest node/machine mismatch');
|
||||
}
|
||||
if ($qa->is_admin || (int)$qa->group_id !== $m['group_id'] ||
|
||||
!str_starts_with($qa->email, 'node-qa-' . $m['key'] . '-') ||
|
||||
!str_ends_with($qa->email, '@example.invalid')) {
|
||||
throw new RuntimeException('Manifest does not identify an isolated QA user');
|
||||
}
|
||||
return [$node, $qa];
|
||||
}
|
||||
|
||||
private static function checkJournal(array $j, array $m): void
|
||||
{
|
||||
if (($j['version'] ?? null) !== 1 || ($j['manifest'] ?? null) !== $m || !is_array($j['operations'] ?? null)) {
|
||||
throw new RuntimeException('Journal/manifest mismatch or unsupported version');
|
||||
}
|
||||
foreach ($j['operations'] as $op) {
|
||||
if (!in_array($op['field'], self::FIELDS[$op['kind']] ?? [], true)) throw new RuntimeException('Invalid journal operation');
|
||||
if ($op['kind']==='node' && $op['id']!==$m['node_id']) throw new RuntimeException('Wrong node in journal');
|
||||
if ($op['kind']==='qa' && $op['id']!==$m['test_user_id']) throw new RuntimeException('Wrong QA in journal');
|
||||
if ($op['kind']==='setting' && $op['id']!=='subscribe_url') throw new RuntimeException('Wrong setting in journal');
|
||||
if ($j['request']['mode']==='append' && !in_array($op['kind'], ['node','qa'], true)) {
|
||||
throw new RuntimeException('Append journal contains shared-resource writes');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static function current(array $op): array
|
||||
{
|
||||
$model = self::object($op['kind'], $op['id']);
|
||||
return [$model, $model !== null, self::value($model, $op['field'])];
|
||||
}
|
||||
|
||||
private static function applyOperation(array $op): void
|
||||
{
|
||||
[$model, $exists, $value] = self::current($op);
|
||||
if ($exists !== $op['before_exists'] || $value !== $op['before']) throw new RuntimeException('Object changed before application');
|
||||
if (!$exists) {
|
||||
if ($op['kind'] !== 'setting') throw new RuntimeException('Missing target');
|
||||
$model = new Setting(['name'=>$op['id']]);
|
||||
}
|
||||
$model->setAttribute($op['field'], $op['after']);
|
||||
// Keep model observers for node full-sync notifications; QA does not send notifications.
|
||||
$op['kind'] === 'qa' ? $model->saveQuietly() : $model->save();
|
||||
}
|
||||
|
||||
private static function flushSettings(array $ops): void
|
||||
{
|
||||
if (array_filter($ops, fn($op) => $op['kind']==='setting')) {
|
||||
// Called after commit; clearing request-local state also makes retry recovery safe.
|
||||
Cache::store('redis')->forget(\App\Support\Setting::CACHE_KEY);
|
||||
app()->forgetInstance(\App\Support\Setting::class);
|
||||
}
|
||||
}
|
||||
|
||||
private static function notifyCommitted(array $m): void
|
||||
{
|
||||
// Explicitly repeat idempotent invalidations after commit. If an observer's
|
||||
// after-commit callback failed, prepared+after recovery must still notify.
|
||||
\App\Services\NodeSyncService::notifyFullSync($m['node_id']);
|
||||
\App\Services\NodeSyncService::notifyMachineNodesChanged($m['machine_id']);
|
||||
}
|
||||
|
||||
private static function verifyEvidence(array $request, array $m, mixed $node): void
|
||||
{
|
||||
$report = readJson($request['verification'] ?? throw new RuntimeException('Missing --verification live QA report'));
|
||||
$at = $report['verified_at'] ?? 0;
|
||||
if (($report['state'] ?? '') !== 'live_verified' || ($report['qa_user_id'] ?? null) !== $m['test_user_id'] ||
|
||||
!is_int($at) || $at > time() || time()-$at > 900) {
|
||||
throw new RuntimeException('Recent live verification for this manifest QA required');
|
||||
}
|
||||
$sni = $node->protocol_settings['tls']['server_name'] ?? null;
|
||||
$targets = array_filter($report['results'] ?? [], fn($row) =>
|
||||
($row['live'] ?? false) === true && ($row['strict_tls'] ?? false) === true &&
|
||||
($row['relay'] ?? null) === $node->host && (int)($row['relay_port'] ?? 0) === (int)$node->port &&
|
||||
($row['sni'] ?? null) === $sni);
|
||||
if (count($targets)!==1) throw new RuntimeException('Live evidence endpoint/SNI mismatch');
|
||||
}
|
||||
|
||||
public static function promote(string $dir, array $request, bool $apply): array
|
||||
{
|
||||
$mode = $request['mode'] ?? '';
|
||||
if (!in_array($mode, ['append','initialize'], true)) throw new RuntimeException('Explicit append/initialize mode required');
|
||||
$request['group'] = isset($request['group']) ? (int)$request['group'] : null;
|
||||
if ($mode==='append' && (!$request['group'] || isset($request['plans']) || isset($request['subscribe-url']) || isset($request['group-name']))) {
|
||||
throw new RuntimeException('Append requires group and forbids plans, subscribe-url and group-name');
|
||||
}
|
||||
if ($mode==='initialize') {
|
||||
$request['plan_ids'] = ids($request['plans'] ?? '');
|
||||
$url = $request['subscribe-url'] ?? '';
|
||||
$parts = parse_url($url);
|
||||
if (!is_array($parts) || ($parts['scheme'] ?? '')!=='https' || empty($parts['host']) || isset($parts['user']) || isset($parts['pass']) || isset($parts['query']) || isset($parts['fragment'])) {
|
||||
throw new RuntimeException('Initialize requires an explicit HTTPS subscribe-url without credentials/query');
|
||||
}
|
||||
}
|
||||
ksort($request);
|
||||
$m = manifest($dir); $path = $dir . '/promotion-journal.json';
|
||||
$result = DB::transaction(function () use ($dir,$request,$mode,$m,$path,$apply) {
|
||||
[$node,$qa] = self::context($m);
|
||||
if (is_file($path)) {
|
||||
$j = readJson($path); self::checkJournal($j,$m);
|
||||
if ($j['request'] !== $request) throw new RuntimeException('Existing operation has different parameters');
|
||||
if (is_file($dir.'/rollback-journal.json')) throw new RuntimeException('Rollback started; use a new reviewed task');
|
||||
$before = $after = true;
|
||||
foreach ($j['operations'] as $op) {
|
||||
[, $exists,$value] = self::current($op);
|
||||
$before = $before && $exists===$op['before_exists'] && $value===$op['before'];
|
||||
$after = $after && $exists && $value===$op['after'];
|
||||
}
|
||||
if (!$after && (!$before || $j['state']!=='prepared')) throw new RuntimeException('Journal conflict; inspect before retrying');
|
||||
if (!$apply) return ['state'=>$after?'already_applied':'preview','journal'=>$j];
|
||||
if (!$after) {
|
||||
self::verifyEvidence($request,$m,$node);
|
||||
if (!$node->is_online || (int)$qa->u+(int)$qa->d<=0 || $qa->banned) throw new RuntimeException('QA evidence no longer ready');
|
||||
foreach ($j['operations'] as $op) self::applyOperation($op);
|
||||
}
|
||||
return ['state'=>'applied','journal'=>$j];
|
||||
}
|
||||
if (is_file($dir.'/promoted.json')) throw new RuntimeException('Legacy promotion exists without new journal; do not adopt automatically');
|
||||
if (!$node->enabled || !$node->show || !$node->is_online || (int)$qa->u + (int)$qa->d <= 0 || $qa->banned) throw new RuntimeException('Fresh online/active QA traffic evidence missing');
|
||||
if (self::value($node,'group_ids') !== [(string)$m['group_id']]) throw new RuntimeException('Node no longer in its isolated stage group');
|
||||
$groupId = $request['group'] ?? $m['group_id'];
|
||||
$group = self::object('group',$groupId) ?? throw new RuntimeException('Target group missing');
|
||||
if ($mode==='append' && $groupId===$m['group_id']) throw new RuntimeException('Append requires an existing formal group');
|
||||
$ops = [self::operation('node',$node,'group_ids',[(string)$groupId]), self::operation('qa',$qa,'banned',true)];
|
||||
if ($mode==='initialize') {
|
||||
$plans = Plan::whereIn('id',$request['plan_ids'])->orderBy('id')->lockForUpdate()->get();
|
||||
if ($plans->count()!==count($request['plan_ids'])) throw new RuntimeException('Plan IDs do not all exist');
|
||||
foreach ($plans as $plan) {
|
||||
if ($plan->group_id !== null && (int)$plan->group_id!==$groupId) throw new RuntimeException('Plan belongs to another group');
|
||||
$ops[] = self::operation('plan',$plan,'group_id',$groupId);
|
||||
}
|
||||
// Only the explicitly selected plans' unassigned users can change in initialize mode.
|
||||
foreach (User::whereIn('plan_id',$request['plan_ids'])->whereNull('group_id')->orderBy('id')->lockForUpdate()->get() as $user) {
|
||||
$ops[] = self::operation('user',$user,'group_id',$groupId);
|
||||
}
|
||||
if (isset($request['group-name'])) $ops[] = self::operation('group',$group,'name',$request['group-name']);
|
||||
$setting = self::object('setting','subscribe_url');
|
||||
$ops[] = self::operation('setting',$setting,'value',$request['subscribe-url'],'subscribe_url');
|
||||
}
|
||||
if ($apply) self::verifyEvidence($request,$m,$node);
|
||||
$j = ['version'=>1,'state'=>'prepared','manifest'=>$m,'request'=>$request,
|
||||
'operations'=>array_values(array_filter($ops)),
|
||||
'node_guard'=>['enabled'=>(bool)$node->enabled,'show'=>(bool)$node->show]];
|
||||
if (!$apply) return ['state'=>'preview','journal'=>$j];
|
||||
writeJson($path,$j); // Durable intent before any DB write; recover prepared+after on retry.
|
||||
foreach ($j['operations'] as $op) self::applyOperation($op);
|
||||
return ['state'=>'applied','journal'=>$j];
|
||||
},1);
|
||||
if ($apply) {
|
||||
self::flushSettings($result['journal']['operations']);
|
||||
self::notifyCommitted($m);
|
||||
$result['journal']['state']='applied'; writeJson($path,$result['journal']);
|
||||
}
|
||||
return $result;
|
||||
}
|
||||
|
||||
public static function rollback(string $dir, bool $apply): array
|
||||
{
|
||||
$m = manifest($dir); $j = readJson($dir.'/promotion-journal.json'); self::checkJournal($j,$m);
|
||||
$rollbackPath=$dir.'/rollback-journal.json';
|
||||
$result=DB::transaction(function () use ($m,$j,$apply,$rollbackPath) {
|
||||
[$node,$qa]=self::context($m);
|
||||
if (!$qa->banned) throw new RuntimeException('QA must remain disabled during rollback');
|
||||
if (is_file($rollbackPath)) {
|
||||
$r=readJson($rollbackPath);
|
||||
if (($r['promotion']??null)!==$j) throw new RuntimeException('Rollback journal mismatch');
|
||||
} else {
|
||||
// Rollback itself has a durable before/after journal, including node containment.
|
||||
$undo=[];
|
||||
foreach (array_reverse($j['operations']) as $op) {
|
||||
if ($op['kind']==='qa') continue; // Never reactivate test users.
|
||||
[$model,$exists,$value]=self::current($op);
|
||||
if (!$exists || $value!==$op['after']) throw new RuntimeException('Rollback conflict: current value differs from recorded after');
|
||||
$undo[]=['kind'=>$op['kind'],'id'=>$op['id'],'field'=>$op['field'],
|
||||
'before_exists'=>true,'before'=>$op['after'],'after'=>$op['before'],
|
||||
'delete'=>!$op['before_exists']];
|
||||
}
|
||||
foreach (['enabled','show'] as $field) {
|
||||
if ((bool)$node->$field!==$j['node_guard'][$field]) throw new RuntimeException('Node containment field changed after promotion');
|
||||
if ($node->$field) $undo[]=['kind'=>'node','id'=>$m['node_id'],'field'=>$field,
|
||||
'before_exists'=>true,'before'=>true,'after'=>false,'delete'=>false];
|
||||
}
|
||||
if (!$qa->banned) throw new RuntimeException('QA reactivated after promotion; resolve conflict before rollback');
|
||||
$r=['version'=>1,'state'=>'prepared','promotion'=>$j,'operations'=>$undo];
|
||||
}
|
||||
$before=$after=true;
|
||||
foreach ($r['operations'] as $op) {
|
||||
// Only rollback containment adds these two node fields.
|
||||
if (!in_array($op['field'],self::FIELDS[$op['kind']]??[],true) && !($op['kind']==='node' && $op['id']===$m['node_id'] && in_array($op['field'],['enabled','show'],true))) {
|
||||
throw new RuntimeException('Invalid rollback field');
|
||||
}
|
||||
[, $exists,$value]=self::current($op);
|
||||
$before=$before && $exists && $value===$op['before'];
|
||||
$after=$after && ($op['delete'] ? !$exists : ($exists && $value===$op['after']));
|
||||
}
|
||||
if (!$after && (!$before || $r['state']!=='prepared')) throw new RuntimeException('Rollback conflicts; no objects restored');
|
||||
if (!$apply) return ['state'=>$after?'already_rolled_back':'preview','journal'=>$r];
|
||||
if (!is_file($rollbackPath)) writeJson($rollbackPath,$r);
|
||||
if (!$after) foreach ($r['operations'] as $op) {
|
||||
[$model]=self::current($op);
|
||||
if ($op['delete']) $model->delete();
|
||||
else { $model->setAttribute($op['field'],$op['after']); $model->save(); }
|
||||
}
|
||||
return ['state'=>'rolled_back','journal'=>$r];
|
||||
},1);
|
||||
if ($apply) {
|
||||
self::flushSettings($j['operations']);
|
||||
self::notifyCommitted($m);
|
||||
$result['journal']['state']='rolled_back';writeJson($rollbackPath,$result['journal']);
|
||||
}
|
||||
return $result;
|
||||
}
|
||||
}
|
||||
66
promote.php
66
promote.php
@ -1,43 +1,25 @@
|
||||
<?php
|
||||
// Run only after real relay, subscription, certificate and traffic checks pass.
|
||||
require getcwd() . '/vendor/autoload.php';
|
||||
$app = require getcwd() . '/bootstrap/app.php';
|
||||
$app->make(Illuminate\Contracts\Console\Kernel::class)->bootstrap();
|
||||
use App\Models\Plan;
|
||||
use App\Models\Server;
|
||||
use App\Models\ServerGroup;
|
||||
use App\Models\User;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
$opts = getopt('', ['backup:', 'plans:', 'group:']);
|
||||
$dir = $opts['backup'] ?? throw new RuntimeException('Missing backup');
|
||||
$ids = array_map('intval', explode(',', $opts['plans'] ?? ''));
|
||||
if (!$ids || in_array(0, $ids, true)) throw new RuntimeException('Explicit plan IDs required');
|
||||
$m = json_decode(file_get_contents("$dir/manifest.json"), true, flags: JSON_THROW_ON_ERROR);
|
||||
$groupId = (int)($opts['group'] ?? $m['group_id']);
|
||||
$node = Server::findOrFail($m['node_id']);
|
||||
$qa = User::findOrFail($m['test_user_id']);
|
||||
if (!$node->is_online || $qa->u + $qa->d <= 0) throw new RuntimeException('Online and traffic evidence missing');
|
||||
if (is_file("$dir/promoted.json")) { echo "Already promoted; verify current state separately\n"; exit; }
|
||||
$plans = Plan::whereIn('id', $ids)->get();
|
||||
if ($plans->count() !== count($ids)) throw new RuntimeException('Plan list mismatch');
|
||||
foreach ($plans as $p) if ($p->group_id !== null && $p->group_id !== $groupId) throw new RuntimeException('Plan already assigned elsewhere');
|
||||
$users = User::whereIn('plan_id', $ids)->whereNull('group_id')->get(['id','group_id','updated_at']);
|
||||
umask(0077);
|
||||
file_put_contents("$dir/before-promotion.json", json_encode([
|
||||
'plans' => $plans->map->getAttributes(), 'user_groups' => $users,
|
||||
'node_groups' => $node->group_ids, 'group' => ServerGroup::findOrFail($groupId)->getAttributes(),
|
||||
'subscribe_url' => admin_setting('subscribe_url'), 'qa_banned' => $qa->banned,
|
||||
], JSON_THROW_ON_ERROR));
|
||||
DB::transaction(function () use ($ids, $users, $node, $qa, $groupId) {
|
||||
$group = ServerGroup::findOrFail($groupId);
|
||||
if (str_starts_with($group->name, 'dpl 接入验收 ')) { $group->name = '基础节点'; $group->save(); }
|
||||
Plan::whereIn('id', $ids)->whereNull('group_id')->update(['group_id' => $groupId]);
|
||||
User::whereIn('id', $users->pluck('id'))->whereNull('group_id')->update(['group_id' => $groupId]);
|
||||
$node->group_ids = [(string)$groupId]; $node->save();
|
||||
$qa->banned = 1; $qa->saveQuietly();
|
||||
admin_setting(['subscribe_url' => 'https://c2.deeplink-subscribe.com']);
|
||||
});
|
||||
$result = ['node_id' => $node->id, 'group_id' => $groupId, 'plan_ids' => $ids,
|
||||
'existing_users_assigned' => $users->count(), 'qa_disabled' => true];
|
||||
file_put_contents("$dir/promoted.json", json_encode($result, JSON_THROW_ON_ERROR));
|
||||
echo json_encode($result) . PHP_EOL;
|
||||
// Default is preview. Apply only in a separately authorized target environment.
|
||||
require __DIR__ . '/operations.php';
|
||||
use DplOnboarding\PanelChange;
|
||||
$opts = getopt('', ['backup:', 'mode:', 'plans:', 'group:', 'subscribe-url:', 'group-name:', 'verification:', 'apply', 'help']);
|
||||
if (isset($opts['help'])) {
|
||||
echo "--backup=DIR --mode=append --group=ID [--verification=QA_REPORT --apply]\n";
|
||||
echo "--backup=DIR --mode=initialize --plans=IDS --subscribe-url=HTTPS_URL [--group=ID] [--group-name=NAME] [--verification=QA_REPORT --apply]\n";
|
||||
exit;
|
||||
}
|
||||
try {
|
||||
$dir = $opts['backup'] ?? throw new RuntimeException('Missing --backup');
|
||||
$apply = array_key_exists('apply', $opts);
|
||||
unset($opts['backup'], $opts['apply']);
|
||||
if (!isset($opts['mode']) || !in_array($opts['mode'], ['append','initialize'], true)) throw new RuntimeException('Explicit --mode required');
|
||||
if (isset($opts['group'])) { DplOnboarding\ids($opts['group']); if (str_contains($opts['group'],',')) throw new RuntimeException('One group required'); }
|
||||
DplOnboarding\boot();
|
||||
$result = DplOnboarding\locked($dir, fn() => PanelChange::promote($dir, $opts, $apply));
|
||||
// Journal is field-scoped and contains no passwords/UUIDs/tokens.
|
||||
echo json_encode($result, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE) . PHP_EOL;
|
||||
} catch (Throwable $e) {
|
||||
$reason=get_class($e)===RuntimeException::class ? $e->getMessage() : "Inspect protected journal/current state";
|
||||
fwrite(STDERR, "Promotion stopped: " . $reason . "\n");
|
||||
exit(1);
|
||||
}
|
||||
|
||||
@ -1,16 +1,32 @@
|
||||
<?php
|
||||
require getcwd() . '/vendor/autoload.php';
|
||||
$app = require getcwd() . '/bootstrap/app.php';
|
||||
$app->make(Illuminate\Contracts\Console\Kernel::class)->bootstrap();
|
||||
$opts = getopt('', ['backup:']);
|
||||
$dir = $opts['backup'] ?? throw new RuntimeException('Missing backup');
|
||||
$m = json_decode(file_get_contents("$dir/manifest.json"), true, flags: JSON_THROW_ON_ERROR);
|
||||
$node = App\Models\Server::findOrFail($m['node_id']);
|
||||
$qa = App\Models\User::findOrFail($m['test_user_id']);
|
||||
echo json_encode([
|
||||
'node_id' => $node->id, 'name' => $node->name, 'is_online' => $node->is_online,
|
||||
'last_check_at' => $node->last_check_at, 'last_push_at' => $node->last_push_at,
|
||||
'qa_upload' => $qa->u, 'qa_download' => $qa->d, 'qa_banned' => $qa->banned,
|
||||
'plans' => App\Models\Plan::orderBy('id')->get(['id', 'group_id']),
|
||||
'existing_users_with_selected_plan_and_null_group' => App\Models\User::whereIn('plan_id', [1,6,7,8])->whereNull('group_id')->count(),
|
||||
], JSON_UNESCAPED_UNICODE) . PHP_EOL;
|
||||
require __DIR__ . '/operations.php';
|
||||
$opts=getopt('', ['backup:', 'plans:', 'group:', 'help']);
|
||||
if (isset($opts['help'])) { echo "--backup=DIR [--plans=ID,ID] [--group=ID]; no implicit plan IDs\n"; exit; }
|
||||
try {
|
||||
$dir=$opts['backup'] ?? throw new RuntimeException('Missing --backup');
|
||||
$ids=isset($opts['plans']) ? DplOnboarding\ids($opts['plans']) : [];
|
||||
$group=null;
|
||||
if (isset($opts['group'])) {
|
||||
$groups=DplOnboarding\ids($opts['group']);
|
||||
if (count($groups)!==1) throw new RuntimeException('One group required');
|
||||
$group=$groups[0];
|
||||
}
|
||||
$m=DplOnboarding\manifest($dir);
|
||||
DplOnboarding\boot();
|
||||
$node=App\Models\Server::findOrFail($m['node_id']);
|
||||
$qa=App\Models\User::findOrFail($m['test_user_id']);
|
||||
if ((int)$node->machine_id!==$m['machine_id']) throw new RuntimeException('Manifest machine mismatch');
|
||||
$plans=App\Models\Plan::whereIn('id',$ids)->orderBy('id')->get(['id','group_id']);
|
||||
if ($plans->count()!==count($ids)) throw new RuntimeException('Some selected plans missing');
|
||||
if ($group!==null) App\Models\ServerGroup::findOrFail($group);
|
||||
$result=[
|
||||
'node_id'=>$node->id, 'name'=>$node->name, 'is_online'=>(bool)$node->is_online,
|
||||
'last_check_at'=>$node->last_check_at, 'last_push_at'=>$node->last_push_at,
|
||||
'group_ids'=>$node->group_ids, 'qa_upload'=>(int)$qa->u, 'qa_download'=>(int)$qa->d,
|
||||
'qa_banned'=>(bool)$qa->banned, 'selected_plan_ids'=>$ids, 'plans'=>$plans->toArray(),
|
||||
'existing_users_with_selected_plan_and_null_group'=>$ids ? App\Models\User::whereIn('plan_id',$ids)->whereNull('group_id')->count() : null,
|
||||
'selected_plans_in_other_group'=>$group===null ? null : $plans->filter(fn($p) => $p->group_id!==null && (int)$p->group_id!==$group)->pluck('id')->all(),
|
||||
'scope'=>'Read-only snapshot; not subscription/traffic/GUI acceptance',
|
||||
];
|
||||
echo json_encode($result,JSON_THROW_ON_ERROR|JSON_UNESCAPED_UNICODE).PHP_EOL;
|
||||
} catch (Throwable $e) { fwrite(STDERR,"Readiness failed; verify parameters and manifest.\n"); exit(1); }
|
||||
|
||||
16
rollback.php
Normal file
16
rollback.php
Normal file
@ -0,0 +1,16 @@
|
||||
<?php
|
||||
// Field-scoped compare-before-restore. Preview unless explicitly --apply.
|
||||
require __DIR__ . '/operations.php';
|
||||
use DplOnboarding\PanelChange;
|
||||
$opts=getopt('', ['backup:', 'apply', 'help']);
|
||||
if (isset($opts['help'])) { echo "--backup=DIR [--apply]; restores new-format promotion journal only\n"; exit; }
|
||||
try {
|
||||
$dir=$opts['backup'] ?? throw new RuntimeException('Missing --backup');
|
||||
DplOnboarding\boot();
|
||||
$result=DplOnboarding\locked($dir, fn() => PanelChange::rollback($dir, array_key_exists('apply',$opts)));
|
||||
echo json_encode($result, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE) . PHP_EOL;
|
||||
} catch (Throwable $e) {
|
||||
$reason=get_class($e)===RuntimeException::class ? $e->getMessage() : "Inspect journal/conflicts";
|
||||
fwrite(STDERR, "Rollback stopped: " . $reason . "\n");
|
||||
exit(1);
|
||||
}
|
||||
179
stage.php
179
stage.php
@ -1,82 +1,109 @@
|
||||
<?php
|
||||
// Run from the production application root. All credential output is private.
|
||||
require getcwd() . '/vendor/autoload.php';
|
||||
$app = require getcwd() . '/bootstrap/app.php';
|
||||
$app->make(Illuminate\Contracts\Console\Kernel::class)->bootstrap();
|
||||
|
||||
use App\Models\Server;
|
||||
use App\Models\ServerGroup;
|
||||
use App\Models\ServerMachine;
|
||||
use App\Models\User;
|
||||
// Run only in an authorized target environment. --help never boots the app.
|
||||
require __DIR__ . '/operations.php';
|
||||
use App\Models\{Server,ServerGroup,ServerMachine,User};
|
||||
use Illuminate\Support\Facades\DB;
|
||||
use Illuminate\Support\Str;
|
||||
|
||||
$opts = getopt('', ['key:', 'name:', 'relay-host:', 'relay-port:', 'listen-port:', 'tls-domain:', 'backup:']);
|
||||
foreach (['key', 'name', 'relay-host', 'relay-port', 'listen-port', 'tls-domain', 'backup'] as $key) {
|
||||
if (empty($opts[$key])) throw new RuntimeException("Missing --$key");
|
||||
}
|
||||
if (!preg_match('/^[a-z0-9-]+$/', $opts['key'])) throw new RuntimeException('Invalid key');
|
||||
foreach (['relay-port', 'listen-port'] as $key) {
|
||||
if (!ctype_digit((string)$opts[$key]) || $opts[$key] < 1024 || $opts[$key] > 65535) throw new RuntimeException('Invalid port');
|
||||
}
|
||||
$dir = rtrim($opts['backup'], '/');
|
||||
umask(0077);
|
||||
if (!is_dir($dir) && !mkdir($dir, 0700, true)) throw new RuntimeException('Backup creation failed');
|
||||
if (is_file("$dir/manifest.json")) {
|
||||
$manifest = json_decode(file_get_contents("$dir/manifest.json"), true, flags: JSON_THROW_ON_ERROR);
|
||||
if ($manifest['key'] !== $opts['key']) throw new RuntimeException('Backup belongs to another node');
|
||||
echo json_encode(['already_staged' => true, 'node_id' => $manifest['node_id']]) . PHP_EOL;
|
||||
$opts=getopt('', ['key:', 'name:', 'relay-host:', 'relay-port:', 'listen-port:', 'tls-domain:', 'panel-url:', 'backup:', 'help']);
|
||||
if (isset($opts['help'])) {
|
||||
echo "--key=KEY --name=NAME --relay-host=HOST --relay-port=PORT --listen-port=PORT --tls-domain=DOMAIN --panel-url=HTTPS_URL --backup=DIR\n";
|
||||
exit;
|
||||
}
|
||||
if (ServerMachine::where('name', 'dpl-' . $opts['key'])->exists() || Server::where('name', $opts['name'])->exists()) {
|
||||
throw new RuntimeException('Existing resources found without this backup manifest');
|
||||
try {
|
||||
foreach (['key','name','relay-host','relay-port','listen-port','tls-domain','panel-url','backup'] as $key) {
|
||||
if (!isset($opts[$key]) || !is_string($opts[$key]) || $opts[$key]==='') throw new RuntimeException('Missing required stage option');
|
||||
}
|
||||
if (!preg_match('/^[a-z0-9-]+$/D',$opts['key'])) throw new RuntimeException('Invalid deployment key');
|
||||
foreach (['relay-port','listen-port'] as $key) {
|
||||
if (!ctype_digit($opts[$key]) || (int)$opts[$key]<1024 || (int)$opts[$key]>65535) throw new RuntimeException('Invalid port');
|
||||
$opts[$key]=(int)$opts[$key];
|
||||
}
|
||||
foreach (['relay-host','tls-domain'] as $key) {
|
||||
if (!preg_match('/^[a-z0-9][a-z0-9.-]*\.[a-z]{2,}$/Di',$opts[$key])) throw new RuntimeException('Explicit hostname required');
|
||||
}
|
||||
$url=parse_url($opts['panel-url']);
|
||||
if (!is_array($url) || ($url['scheme']??'')!=='https' || empty($url['host']) || isset($url['user']) || isset($url['pass']) || isset($url['query']) || isset($url['fragment'])) {
|
||||
throw new RuntimeException('Explicit HTTPS panel URL without credentials/query required');
|
||||
}
|
||||
$dir=rtrim($opts['backup'],'/');
|
||||
if (!str_starts_with($dir,'/') || is_link($dir)) throw new RuntimeException('Absolute private backup directory required');
|
||||
umask(0077);
|
||||
if (!is_dir($dir) && !mkdir($dir,0700,true)) throw new RuntimeException('Cannot create backup directory');
|
||||
if ((fileperms($dir)&0077)!==0) throw new RuntimeException('Backup directory must be private');
|
||||
$parameters=$opts;unset($parameters['backup']);ksort($parameters);
|
||||
DplOnboarding\boot();
|
||||
$m=DplOnboarding\locked($dir,function () use ($dir,$parameters) {
|
||||
$path=$dir.'/manifest.json';
|
||||
if (is_file($path)) {
|
||||
$m=DplOnboarding\manifest($dir);
|
||||
if (($m['parameters']??null)!==$parameters) throw new RuntimeException('Existing manifest has different/legacy parameters; no automatic adoption');
|
||||
$node=Server::findOrFail($m['node_id']);$qa=User::findOrFail($m['test_user_id']);
|
||||
$machine=ServerMachine::findOrFail($m['machine_id']);
|
||||
if ((int)$node->machine_id!==$machine->id || $machine->name!=='dpl-'.$parameters['key'] ||
|
||||
$node->host!==$parameters['relay-host'] || (int)$node->port!==$parameters['relay-port'] ||
|
||||
(int)$node->server_port!==$parameters['listen-port'] || $node->type!=='anytls' ||
|
||||
($node->protocol_settings['tls']['server_name']??null)!==$parameters['tls-domain'] ||
|
||||
(int)$qa->group_id!==$m['group_id']) throw new RuntimeException('Staged resources drifted');
|
||||
if (is_file($dir.'/promotion-journal.json') || is_file($dir.'/promoted.json')) throw new RuntimeException('Node already promoted; staging cannot regenerate credentials');
|
||||
return $m;
|
||||
}
|
||||
return DB::transaction(function () use ($parameters,$path,$dir) {
|
||||
if (ServerMachine::where('name','dpl-'.$parameters['key'])->exists() || Server::where('name',$parameters['name'])->exists() ||
|
||||
Server::where('host',$parameters['relay-host'])->where('port',(string)$parameters['relay-port'])->exists() ||
|
||||
ServerGroup::where('name','dpl 接入验收 '.$parameters['key'])->exists()) {
|
||||
throw new RuntimeException('Existing resource/relay tuple without matching manifest; refusing takeover');
|
||||
}
|
||||
DplOnboarding\writeJson($dir.'/before.json',[
|
||||
'server_ids'=>Server::orderBy('id')->pluck('id')->all(),
|
||||
'machine_ids'=>ServerMachine::orderBy('id')->pluck('id')->all(),
|
||||
'group_ids'=>ServerGroup::orderBy('id')->pluck('id')->all(),
|
||||
]);
|
||||
$group=new ServerGroup();$group->name='dpl 接入验收 '.$parameters['key'];$group->save();
|
||||
$machine=ServerMachine::create(['name'=>'dpl-'.$parameters['key'],
|
||||
'notes'=>'Independent AnyTLS service; existing XrayR preserved','is_active'=>true,'token'=>ServerMachine::generateToken()]);
|
||||
$node=Server::create([
|
||||
'name'=>$parameters['name'],'type'=>'anytls','host'=>$parameters['relay-host'],
|
||||
'port'=>(string)$parameters['relay-port'],'server_port'=>(string)$parameters['listen-port'],
|
||||
'group_ids'=>[(string)$group->id],'route_ids'=>[],'tags'=>[],
|
||||
'rate'=>1,'show'=>true,'enabled'=>true,'machine_id'=>$machine->id,'sort'=>1,
|
||||
'protocol_settings'=>['tls'=>['server_name'=>$parameters['tls-domain'],'allow_insecure'=>false]],
|
||||
'cert_config'=>['cert_mode'=>'file','domain'=>$parameters['tls-domain'],
|
||||
'cert_file'=>'/etc/letsencrypt/live/'.$parameters['tls-domain'].'/fullchain.pem',
|
||||
'key_file'=>'/etc/letsencrypt/live/'.$parameters['tls-domain'].'/privkey.pem'],
|
||||
]);
|
||||
$qa=new User([
|
||||
'email'=>'node-qa-'.$parameters['key'].'-'.Str::lower(Str::random(8)).'@example.invalid',
|
||||
'password'=>password_hash(Str::random(40),PASSWORD_BCRYPT),'uuid'=>(string)Str::uuid(),'token'=>Str::random(32),
|
||||
'group_id'=>$group->id,'transfer_enable'=>104857600,'expired_at'=>time()+86400,'banned'=>0,'is_admin'=>0,
|
||||
'remind_expire'=>0,'remind_traffic'=>0,'device_limit'=>2,
|
||||
]);$qa->saveQuietly();
|
||||
$m=['key'=>$parameters['key'],'node_id'=>$node->id,'machine_id'=>$machine->id,'group_id'=>$group->id,
|
||||
'test_user_id'=>$qa->id,'parameters'=>$parameters];
|
||||
// Save identities before commit. A failed/unknown commit must be inspected,
|
||||
// never repeated into new resources merely because a response was lost.
|
||||
DplOnboarding\writeJson($path,$m);
|
||||
return $m;
|
||||
},1);
|
||||
});
|
||||
// Recovery can regenerate missing transport files from the same staged objects.
|
||||
$machine=ServerMachine::findOrFail($m['machine_id']);$qa=User::findOrFail($m['test_user_id']);
|
||||
$panel=json_encode($parameters['panel-url'],JSON_THROW_ON_ERROR|JSON_UNESCAPED_SLASHES);
|
||||
$config="panel:\n url: $panel\nmachine:\n machine_id: {$machine->id}\n token_env: DPL_MACHINE_TOKEN\nkernel:\n type: singbox\n config_dir: /var/lib/dpl-node/kernel\nlog:\n level: info\nruntime:\n gomemlimit: 384MiB\n gogc: 75\nnode:\n push_interval: 30\n pull_interval: 30\n";
|
||||
foreach (['node-config.yml'=>$config,'node-secret.env'=>'DPL_MACHINE_TOKEN='.$machine->token."\n"] as $file=>$body) {
|
||||
$path=$dir.'/'.$file;
|
||||
if (is_link($path)) throw new RuntimeException('Refusing private transport symlink');
|
||||
$tmp=tempnam($dir,'.private-');if ($tmp===false) throw new RuntimeException('Private file failed');
|
||||
chmod($tmp,0600);
|
||||
if (file_put_contents($tmp,$body)!==strlen($body) || !rename($tmp,$path)) {
|
||||
if (is_file($tmp)) unlink($tmp);
|
||||
throw new RuntimeException('Private file failed');
|
||||
}
|
||||
}
|
||||
DplOnboarding\writeJson($dir.'/qa-private.json',['user_id'=>$qa->id,'uuid'=>$qa->uuid,'subscribe_token'=>$qa->token,
|
||||
'host'=>$parameters['relay-host'],'port'=>$parameters['relay-port'],'sni'=>$parameters['tls-domain']]);
|
||||
echo json_encode(['key'=>$m['key'],'node_id'=>$m['node_id'],'machine_id'=>$m['machine_id'],
|
||||
'group_id'=>$m['group_id'],'test_user_id'=>$m['test_user_id']],JSON_THROW_ON_ERROR).PHP_EOL;
|
||||
} catch (Throwable $e) {
|
||||
$reason=get_class($e)===RuntimeException::class ? $e->getMessage() : 'Inspect manifest/DB state before retrying';
|
||||
fwrite(STDERR,'Staging stopped: '.$reason.PHP_EOL);exit(1);
|
||||
}
|
||||
file_put_contents("$dir/before.json", json_encode([
|
||||
'servers' => DB::table('v2_server')->get(),
|
||||
'groups' => DB::table('v2_server_group')->get(),
|
||||
'machines' => DB::table('v2_server_machine')->get(),
|
||||
], JSON_THROW_ON_ERROR));
|
||||
|
||||
$manifest = DB::transaction(function () use ($opts) {
|
||||
$group = new ServerGroup();
|
||||
$group->name = 'dpl 接入验收 ' . $opts['key'];
|
||||
$group->save();
|
||||
$machine = ServerMachine::create([
|
||||
'name' => 'dpl-' . $opts['key'], 'notes' => 'Independent AnyTLS service; existing XrayR preserved',
|
||||
'is_active' => true, 'token' => ServerMachine::generateToken(),
|
||||
]);
|
||||
$node = Server::create([
|
||||
'name' => $opts['name'], 'type' => 'anytls', 'host' => $opts['relay-host'],
|
||||
'port' => (string)$opts['relay-port'], 'server_port' => (string)$opts['listen-port'],
|
||||
'group_ids' => [(string)$group->id], 'route_ids' => [], 'tags' => [],
|
||||
'rate' => 1, 'show' => true, 'enabled' => true, 'machine_id' => $machine->id, 'sort' => 1,
|
||||
'protocol_settings' => ['tls' => ['server_name' => $opts['tls-domain'], 'allow_insecure' => false]],
|
||||
'cert_config' => [
|
||||
'cert_mode' => 'file', 'domain' => $opts['tls-domain'],
|
||||
'cert_file' => '/etc/letsencrypt/live/' . $opts['tls-domain'] . '/fullchain.pem',
|
||||
'key_file' => '/etc/letsencrypt/live/' . $opts['tls-domain'] . '/privkey.pem',
|
||||
],
|
||||
]);
|
||||
$user = new User([
|
||||
'email' => 'node-qa-' . $opts['key'] . '-' . Str::lower(Str::random(8)) . '@example.invalid',
|
||||
'password' => password_hash(Str::random(40), PASSWORD_BCRYPT),
|
||||
'uuid' => (string)Str::uuid(), 'token' => Str::random(32),
|
||||
'group_id' => $group->id, 'transfer_enable' => 104857600,
|
||||
'expired_at' => time() + 86400, 'banned' => 0, 'is_admin' => 0,
|
||||
'remind_expire' => 0, 'remind_traffic' => 0, 'device_limit' => 2,
|
||||
]);
|
||||
$user->saveQuietly();
|
||||
return ['key' => $opts['key'], 'node_id' => $node->id, 'machine_id' => $machine->id,
|
||||
'group_id' => $group->id, 'test_user_id' => $user->id];
|
||||
});
|
||||
file_put_contents("$dir/manifest.json", json_encode($manifest, JSON_THROW_ON_ERROR));
|
||||
$machine = ServerMachine::findOrFail($manifest['machine_id']);
|
||||
$user = User::findOrFail($manifest['test_user_id']);
|
||||
$config = "panel:\n url: \"https://dp-link.net\"\nmachine:\n machine_id: {$machine->id}\n token_env: DPL_MACHINE_TOKEN\nkernel:\n type: singbox\n config_dir: /var/lib/dpl-node/kernel\nlog:\n level: info\nruntime:\n gomemlimit: 384MiB\n gogc: 75\nnode:\n push_interval: 30\n pull_interval: 30\n";
|
||||
file_put_contents("$dir/node-config.yml", $config);
|
||||
file_put_contents("$dir/node-secret.env", 'DPL_MACHINE_TOKEN=' . $machine->token . "\n");
|
||||
file_put_contents("$dir/qa-private.json", json_encode([
|
||||
'user_id' => $user->id, 'uuid' => $user->uuid, 'subscribe_token' => $user->token,
|
||||
'host' => $opts['relay-host'], 'port' => (int)$opts['relay-port'], 'sni' => $opts['tls-domain'],
|
||||
], JSON_THROW_ON_ERROR));
|
||||
echo json_encode($manifest, JSON_UNESCAPED_UNICODE) . PHP_EOL;
|
||||
|
||||
284
verify-node.py
284
verify-node.py
@ -1,72 +1,240 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Run an isolated SOCKS client with strict TLS; never print QA credentials."""
|
||||
"""Validate private subscriptions; network execution requires explicit --run.
|
||||
|
||||
For a formal multi-node subscription use a private --expectations JSON array:
|
||||
[{"tag": "exact outbound tag", "host": "relay.example", "port": 12345,
|
||||
"sni": "node.example", "uuid": "private password", "expected_exit": "192.0.2.1",
|
||||
"expected_country": "JP"}]. Never print credentials or the subscription.
|
||||
"""
|
||||
import argparse
|
||||
import copy
|
||||
import ipaddress
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--qa', required=True)
|
||||
p.add_argument('--binary', required=True)
|
||||
p.add_argument('--expected-exit', required=True)
|
||||
p.add_argument('--port', type=int, default=19080)
|
||||
p.add_argument('--subscription')
|
||||
a = p.parse_args()
|
||||
q = json.loads(pathlib.Path(a.qa).read_text())
|
||||
c = {'log': {'level': 'error'}, 'dns': {'servers': [{'type': 'local', 'tag': 'local'}]},
|
||||
'inbounds': [{'type': 'mixed', 'listen': '127.0.0.1', 'listen_port': a.port}],
|
||||
'outbounds': [{'type': 'anytls', 'tag': 'qa', 'server': q['host'], 'server_port': q['port'],
|
||||
'password': q['uuid'], 'tls': {'enabled': True, 'server_name': q['sni'], 'insecure': False}}],
|
||||
'route': {'default_domain_resolver': 'local', 'final': 'qa'}}
|
||||
if a.subscription:
|
||||
c = json.loads(pathlib.Path(a.subscription).read_text())
|
||||
nodes = [x for x in c['outbounds'] if x.get('type') == 'anytls']
|
||||
assert len(nodes) == 1 and nodes[0]['tls'].get('insecure') is False
|
||||
assert nodes[0]['server'] == q['host'] and nodes[0]['server_port'] == q['port']
|
||||
assert nodes[0]['password'] == q['uuid']
|
||||
c['inbounds'] = [{'type': 'mixed', 'listen': '127.0.0.1', 'listen_port': a.port}]
|
||||
c['log'] = {'level': 'error'}
|
||||
os.umask(0o077)
|
||||
with tempfile.TemporaryDirectory(prefix='dpl-client-') as d:
|
||||
config = pathlib.Path(d) / 'client.json'
|
||||
config.write_text(json.dumps(c))
|
||||
env = dict(os.environ, GOMEMLIMIT='64MiB', GOGC='75')
|
||||
with (pathlib.Path(d) / 'client.log').open('wb') as log:
|
||||
proc = subprocess.Popen([a.binary, 'run', '-c', str(config)], cwd=d, env=env, stdout=log, stderr=log)
|
||||
|
||||
class VerificationError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def require(condition, message):
|
||||
if not condition:
|
||||
raise VerificationError(message)
|
||||
|
||||
|
||||
def read_private(path):
|
||||
p = Path(path)
|
||||
require(p.is_file() and not p.is_symlink(), 'Private input must be a regular file')
|
||||
require(p.stat().st_mode & 0o077 == 0, 'Private input must have no group/other permissions')
|
||||
return json.loads(p.read_text())
|
||||
|
||||
|
||||
def validate(config, expectations, exact=True):
|
||||
require(isinstance(config, dict) and isinstance(config.get('outbounds'), list), 'Missing outbounds')
|
||||
all_tags = [o.get('tag') for o in config['outbounds'] if isinstance(o, dict)]
|
||||
require(len(all_tags) == len(config['outbounds']) and all(isinstance(t, str) and t for t in all_tags), 'Each outbound needs a tag')
|
||||
require(len(set(all_tags)) == len(all_tags), 'Duplicate outbound tags')
|
||||
nodes = {o['tag']: o for o in config['outbounds'] if o.get('type') == 'anytls'}
|
||||
require(isinstance(expectations, list) and expectations, 'Nonempty expectations required')
|
||||
selected = []
|
||||
for spec in expectations:
|
||||
require(isinstance(spec, dict), 'Invalid expectation')
|
||||
for key in ['tag', 'host', 'sni', 'uuid', 'expected_exit', 'expected_country']:
|
||||
require(isinstance(spec.get(key), str) and bool(spec[key]), 'Expectation missing required field')
|
||||
require(isinstance(spec.get('port'), int) and not isinstance(spec['port'], bool) and 1 <= spec['port'] <= 65535, 'Invalid relay port')
|
||||
try:
|
||||
for _ in range(50):
|
||||
assert proc.poll() is None, 'Client exited during startup'
|
||||
try:
|
||||
with socket.create_connection(('127.0.0.1', a.port), 0.1):
|
||||
break
|
||||
except OSError:
|
||||
time.sleep(0.1)
|
||||
command = ['curl', '-q', '--fail', '--silent', '--show-error', '--max-time', '30',
|
||||
'--socks5-hostname', f'127.0.0.1:{a.port}']
|
||||
result = subprocess.run(command + ['https://api.ipify.org'], capture_output=True, text=True, check=True)
|
||||
ip = result.stdout.strip()
|
||||
assert ip == a.expected_exit, 'Unexpected exit IP'
|
||||
result = subprocess.run(command + ['https://www.cloudflare.com/cdn-cgi/trace'], capture_output=True, text=True, check=True)
|
||||
trace = dict(line.split('=', 1) for line in result.stdout.splitlines() if '=' in line)
|
||||
assert trace.get('ip') == a.expected_exit, 'Second exit check mismatch'
|
||||
print(json.dumps({'anytls': True, 'relay': q['host'], 'port': q['port'], 'strict_tls': True,
|
||||
'exit_ip': ip, 'country': trace.get('loc'), 'https_tls': trace.get('tls')}, ensure_ascii=False))
|
||||
ipaddress.ip_address(spec['expected_exit'])
|
||||
except ValueError as exc:
|
||||
raise VerificationError('Invalid expected exit IP') from exc
|
||||
require(len(spec['expected_country']) == 2 and spec['expected_country'].isalpha() and spec['expected_country'].isupper(), 'Expected country must be two uppercase letters')
|
||||
tag = spec['tag']
|
||||
require(tag in nodes and tag not in selected, 'Missing/duplicate selected node')
|
||||
node = nodes[tag]
|
||||
tls = node.get('tls') or {}
|
||||
require(tls.get('enabled') is True and tls.get('insecure') is False, 'Strict TLS must be explicit')
|
||||
require(node.get('server') == spec['host'] and node.get('server_port') == spec['port'], 'Relay endpoint mismatch')
|
||||
require(tls.get('server_name') == spec['sni'], 'SNI mismatch')
|
||||
require(node.get('password') == spec['uuid'], 'Subscription password mismatch')
|
||||
selected.append(tag)
|
||||
if exact:
|
||||
require(set(selected) == set(nodes), 'Expectations must cover every AnyTLS node exactly')
|
||||
return selected
|
||||
|
||||
|
||||
def probe_config(config, spec, port):
|
||||
result = copy.deepcopy(config)
|
||||
result['inbounds'] = [{'type': 'mixed', 'listen': '127.0.0.1', 'listen_port': port}]
|
||||
result['log'] = {'level': 'error'}
|
||||
# Preserve DNS, remote rulesets and all original outbounds. Only the two
|
||||
# diagnostic destinations are pinned; ordinary subscription routing remains.
|
||||
route = result.setdefault('route', {})
|
||||
rules = route.setdefault('rules', [])
|
||||
require(isinstance(rules, list), 'Invalid route rules')
|
||||
rules.insert(0, {'domain': ['api.ipify.org', 'www.cloudflare.com'],
|
||||
'action': 'route', 'outbound': spec['tag']})
|
||||
return result
|
||||
|
||||
|
||||
def stop(proc):
|
||||
if proc is None or proc.poll() is not None:
|
||||
return
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
proc.wait(timeout=5)
|
||||
|
||||
|
||||
def run_node(config, spec, binary, port, diagnostic_dir):
|
||||
require(1024 <= port <= 65535, 'Invalid local SOCKS port')
|
||||
with socket.socket() as free:
|
||||
try:
|
||||
free.bind(('127.0.0.1', port))
|
||||
except OSError as exc:
|
||||
raise VerificationError('Local SOCKS port is occupied') from exc
|
||||
with tempfile.TemporaryDirectory(prefix='dpl-client-', dir=diagnostic_dir) as tmp:
|
||||
temporary = Path(tmp)
|
||||
path = temporary / 'client.json'
|
||||
path.write_text(json.dumps(probe_config(config, spec, port)))
|
||||
path.chmod(0o600)
|
||||
env = dict(os.environ, GOMEMLIMIT='64MiB', GOGC='75')
|
||||
proc = None
|
||||
log_path = temporary / 'client.log'
|
||||
try:
|
||||
with log_path.open('wb') as log:
|
||||
subprocess.run([binary, 'check', '-c', str(path)], env=env,
|
||||
cwd=tmp, stdout=log, stderr=log, check=True, timeout=30)
|
||||
proc = subprocess.Popen([binary, 'run', '-c', str(path)], env=env,
|
||||
cwd=tmp, stdout=log, stderr=log)
|
||||
ready = False
|
||||
for _ in range(100):
|
||||
require(proc.poll() is None, 'Client exited during startup')
|
||||
try:
|
||||
with socket.create_connection(('127.0.0.1', port), 0.1):
|
||||
ready = True
|
||||
break
|
||||
except OSError:
|
||||
time.sleep(0.1)
|
||||
require(ready, 'Client startup timed out')
|
||||
command = ['curl', '-q', '--fail', '--silent', '--show-error', '--max-time', '30',
|
||||
'--socks5-hostname', f'127.0.0.1:{port}', '--noproxy', '']
|
||||
ip = subprocess.run(command + ['https://api.ipify.org'], capture_output=True,
|
||||
text=True, check=True, timeout=35).stdout.strip()
|
||||
require(ipaddress.ip_address(ip) == ipaddress.ip_address(spec['expected_exit']), 'Unexpected exit IP')
|
||||
response = subprocess.run(command + ['https://www.cloudflare.com/cdn-cgi/trace'],
|
||||
capture_output=True, text=True, check=True, timeout=35)
|
||||
trace = dict(line.split('=', 1) for line in response.stdout.splitlines() if '=' in line)
|
||||
require(trace.get('ip') == ip, 'Second exit check mismatch')
|
||||
require(trace.get('loc') == spec['expected_country'], 'Exit country mismatch')
|
||||
require(trace.get('tls', '').startswith('TLSv'), 'Target HTTPS TLS not observed')
|
||||
return {'tag': spec['tag'], 'strict_tls': True, 'exit_ip': ip,
|
||||
'country': trace['loc'], 'https_tls': trace['tls'], 'live': True,
|
||||
'relay': spec['host'], 'relay_port': spec['port'], 'sni': spec['sni']}
|
||||
except Exception:
|
||||
log.flush()
|
||||
failure = pathlib.Path(a.qa).parent / 'client-failure.log'
|
||||
shutil.copyfile(pathlib.Path(d) / 'client.log', failure)
|
||||
failure.chmod(0o600)
|
||||
print(f'Client failed; protected diagnostic log: {failure}')
|
||||
# Logs may contain private config context; retain only in private task directory.
|
||||
stop(proc)
|
||||
if log_path.exists():
|
||||
fd, name = tempfile.mkstemp(prefix='client-failure-', suffix='.log', dir=diagnostic_dir)
|
||||
os.close(fd)
|
||||
failure = Path(name)
|
||||
shutil.copyfile(log_path, failure)
|
||||
failure.chmod(0o600)
|
||||
raise
|
||||
finally:
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
proc.wait()
|
||||
stop(proc)
|
||||
|
||||
|
||||
def write_report(path, report):
|
||||
destination = Path(path)
|
||||
require(destination.parent.is_dir() and destination.parent.stat().st_mode & 0o077 == 0,
|
||||
'Report directory must be private')
|
||||
require(not destination.is_symlink(), 'Refusing report symlink')
|
||||
with tempfile.NamedTemporaryFile(mode='w', dir=destination.parent, delete=False) as stream:
|
||||
temporary = Path(stream.name)
|
||||
try:
|
||||
os.chmod(temporary, 0o600)
|
||||
json.dump(report, stream, ensure_ascii=False)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temporary, destination)
|
||||
finally:
|
||||
if temporary.exists():
|
||||
temporary.unlink()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
sources = parser.add_mutually_exclusive_group(required=True)
|
||||
sources.add_argument('--qa')
|
||||
sources.add_argument('--expectations')
|
||||
parser.add_argument('--subscription', required=True)
|
||||
parser.add_argument('--binary')
|
||||
parser.add_argument('--expected-exit')
|
||||
parser.add_argument('--expected-country')
|
||||
parser.add_argument('--target-tag')
|
||||
parser.add_argument('--report', help='Private JSON result file; required for --run')
|
||||
parser.add_argument('--port', type=int, default=19080)
|
||||
execution = parser.add_mutually_exclusive_group()
|
||||
execution.add_argument('--run', action='store_true')
|
||||
execution.add_argument('--validate-only', action='store_true')
|
||||
args = parser.parse_args()
|
||||
os.umask(0o077)
|
||||
try:
|
||||
require(not args.run or args.report, '--run requires a private --report destination')
|
||||
if args.report:
|
||||
destination = Path(args.report)
|
||||
inputs = [args.subscription, args.qa, args.expectations]
|
||||
require(all(destination.resolve() != Path(item).resolve() for item in inputs if item), 'Report must not overwrite a private input')
|
||||
require(destination.parent.is_dir() and destination.parent.stat().st_mode & 0o077 == 0 and not destination.is_symlink(), 'Private regular report destination required')
|
||||
if args.report:
|
||||
write_report(args.report, {'state': 'verification_pending', 'verified_at': None})
|
||||
config = read_private(args.subscription)
|
||||
qa_user_id = None
|
||||
source = args.expectations or args.qa
|
||||
if args.expectations:
|
||||
require(not any([args.expected_exit, args.expected_country, args.target_tag]), 'Put each node expectation in the private expectations file')
|
||||
specs = read_private(args.expectations)
|
||||
else:
|
||||
qa = read_private(args.qa)
|
||||
qa_user_id = qa.get('user_id')
|
||||
require(args.expected_exit and args.expected_country, 'QA requires explicit expected-exit and expected-country')
|
||||
candidates = [n for n in config.get('outbounds', []) if n.get('type') == 'anytls'
|
||||
and n.get('server') == qa.get('host') and n.get('server_port') == qa.get('port')
|
||||
and (not args.target_tag or n.get('tag') == args.target_tag)]
|
||||
require(len(candidates) == 1, 'QA target ambiguous/missing; supply exact target-tag')
|
||||
specs = [{'tag': candidates[0]['tag'], 'host': qa.get('host'), 'port': qa.get('port'),
|
||||
'sni': qa.get('sni'), 'uuid': qa.get('uuid'), 'expected_exit': args.expected_exit,
|
||||
'expected_country': args.expected_country}]
|
||||
tags = validate(config, specs, exact=bool(args.expectations))
|
||||
results = []
|
||||
if args.run:
|
||||
require(args.binary and Path(args.binary).is_file(), 'Explicit existing client binary required')
|
||||
binary = str(Path(args.binary).resolve())
|
||||
folder = Path(source).resolve().parent
|
||||
require(folder.stat().st_mode & 0o077 == 0, 'Live diagnostics directory must be private')
|
||||
for spec in specs:
|
||||
results.append(run_node(config, spec, binary, args.port, str(folder)))
|
||||
report = {'state': 'live_verified' if args.run else 'structure_validated',
|
||||
'verified_at': int(time.time()) if args.run else None,
|
||||
'qa_user_id': qa_user_id, 'selected_nodes': tags, 'results': results,
|
||||
'scope': 'Full subscription retained; only diagnostic destination routes pinned. No traffic/account/GUI acceptance implied.'}
|
||||
if args.report:
|
||||
write_report(args.report, report)
|
||||
print(json.dumps(report, ensure_ascii=False))
|
||||
return 0
|
||||
except (Exception, KeyboardInterrupt) as exc:
|
||||
# No traceback or subprocess stderr: either can contain credentials.
|
||||
message = str(exc) if isinstance(exc, VerificationError) else 'Verification failed; inspect protected diagnostics'
|
||||
print(json.dumps({'state': 'failed', 'reason': message}), file=__import__('sys').stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
|
||||
Loading…
Reference in New Issue
Block a user