Files
hy2-panel/lib/templates/relay-agent.py
T

122 lines
4.2 KiB
Python

#!/usr/bin/env python3
"""Outbound-only management for a dedicated Realm instance."""
import fcntl
import json
import os
from pathlib import Path
import socket
import subprocess
import time
import urllib.request
import urllib.error
ROOT = Path('/etc/hy2-panel-relay')
UNIT = 'hy2-panel-relay.service'
def core(command):
subprocess.run(['systemctl', command, UNIT], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
def pid():
return int(subprocess.check_output(['systemctl', 'show', UNIT, '-p', 'MainPID', '--value'], text=True).strip())
def active():
return subprocess.run(['systemctl', 'is-active', '--quiet', UNIT]).returncode == 0
def verify_listeners(config):
process = pid()
if not process:
return False
inodes = set()
for fd in Path('/proc/' + str(process) + '/fd').iterdir():
try:
link = os.readlink(fd)
if link.startswith('socket:['):
inodes.add(link[8:-1])
except OSError:
pass
ports = set()
for network in ('tcp', 'udp'):
for line in Path('/proc/net/' + network).read_text().splitlines()[1:]:
fields = line.split()
if fields[9] in inodes and (network != 'tcp' or fields[3] == '0A'):
ports.add((network, int(fields[1].split(':')[1], 16)))
return all(('udp' if endpoint['network']['use_udp'] else 'tcp', int(endpoint['listen'].rsplit(':', 1)[1])) in ports for endpoint in config['endpoints'])
def write_config(config):
staged = ROOT / 'next.json'
with open(staged, 'w') as handle:
json.dump(config, handle)
handle.flush()
os.fsync(handle.fileno())
os.replace(staged, ROOT / 'config.json')
def apply(config):
core('stop')
sockets = []
try:
for endpoint in config['endpoints']:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM if endpoint['network']['use_udp'] else socket.SOCK_STREAM)
sockets.append(sock)
# SO_REUSEADDR allows a stopped TCP listener's TIME_WAIT connections.
if not endpoint['network']['use_udp']:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('0.0.0.0', int(endpoint['listen'].rsplit(':', 1)[1])))
except OSError:
return 'port_conflict'
finally:
for sock in sockets:
sock.close()
write_config(config)
if config['endpoints']:
core('start')
for _ in range(20):
time.sleep(0.2)
if active() and verify_listeners(config):
return ''
core('stop')
return 'apply_failed'
return ''
def run():
lock = open(ROOT / 'agent.lock', 'w')
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
credentials = json.loads((ROOT / 'agent.json').read_text())
applied = ''
error = ''
deadline = 0
core('stop')
while True:
try:
request = urllib.request.Request(credentials['panel'] + '/api/relay/' + credentials['id'] + '/sync',
data=json.dumps({'appliedHash': applied, 'active': active(), 'error': error}).encode(),
headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + credentials['token']})
with urllib.request.urlopen(request, timeout=10) as response:
desired = json.load(response)
if desired['hash'] != applied or (desired['config']['endpoints'] and not active()):
error = apply(desired['config'])
applied = desired['hash'] if not error else ''
deadline = time.monotonic() + min(90, desired['leaseSeconds'])
print('同步成功' if not error else error, flush=True)
except Exception as exc:
print('同步失败:' + type(exc).__name__, flush=True)
if isinstance(exc, urllib.error.HTTPError) and exc.code in (400, 401, 403, 404):
deadline = 0
if time.monotonic() >= deadline:
core('stop')
applied = ''
time.sleep(max(1, min(15, deadline - time.monotonic())) if deadline else 5)
if __name__ == '__main__':
os.umask(0o077)
try:
run()
finally:
core('stop')