From b8d10fac963d63825c97d36d41ff7f525116bbfd Mon Sep 17 00:00:00 2001 From: hellcat Date: Sat, 17 Oct 2026 08:30:10 +0800 Subject: [PATCH] Add reviewed append-only node operations and batch coordination --- batch.py | 368 +++++++++++++++++++++++++++++++++++++++++++++ command-adapter.py | 66 ++++++++ operations.php | 332 ++++++++++++++++++++++++++++++++++++++++ promote.php | 66 +++----- readiness.php | 46 ++++-- rollback.php | 16 ++ stage.php | 179 ++++++++++++---------- verify-node.py | 284 +++++++++++++++++++++++++++------- 8 files changed, 1166 insertions(+), 191 deletions(-) create mode 100644 batch.py create mode 100644 command-adapter.py create mode 100644 operations.php create mode 100644 rollback.php diff --git a/batch.py b/batch.py new file mode 100644 index 0000000..baeb5de --- /dev/null +++ b/batch.py @@ -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()) diff --git a/command-adapter.py b/command-adapter.py new file mode 100644 index 0000000..9c54328 --- /dev/null +++ b/command-adapter.py @@ -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()) diff --git a/operations.php b/operations.php new file mode 100644 index 0000000..54d1d0f --- /dev/null +++ b/operations.php @@ -0,0 +1,332 @@ +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; + } +} diff --git a/promote.php b/promote.php index 453a801..10299d1 100644 --- a/promote.php +++ b/promote.php @@ -1,43 +1,25 @@ 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); +} diff --git a/readiness.php b/readiness.php index 42650e4..58fb0eb 100644 --- a/readiness.php +++ b/readiness.php @@ -1,16 +1,32 @@ 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); } diff --git a/rollback.php b/rollback.php new file mode 100644 index 0000000..b8c5657 --- /dev/null +++ b/rollback.php @@ -0,0 +1,16 @@ + 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); +} diff --git a/stage.php b/stage.php index e3de6a2..6202553 100644 --- a/stage.php +++ b/stage.php @@ -1,82 +1,109 @@ 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; diff --git a/verify-node.py b/verify-node.py index 0c99d86..e08a948 100644 --- a/verify-node.py +++ b/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())