#!/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')