#!/usr/bin/env python3
"""Temporary Linux runner. Standard library only; interactive approval is mandatory."""
import argparse
import codecs
import ipaddress
import json
import os
import selectors
import signal
import socket
import subprocess
import sys
import threading
import time
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

VERSION = '0.1.0'
DEFAULT_HUB = 'https://help.5itpro.com'
OUTPUT_LIMIT = 65536


class RunnerError(Exception):
    pass


def safe_text(text):
    return ''.join(char if char in '\n\t' or unicodedata.category(char) not in ('Cc', 'Cf')
                   else f'\\u{ord(char):04x}' for char in str(text))


def normalize_code(value):
    value = value.strip().strip('"')
    compact = ''.join(c for c in value if not c.isspace() and c not in '-–—')
    return compact.upper() if len(compact) == 10 else value


def validate_url(value):
    value = value.strip().rstrip('/')
    parts = urllib.parse.urlsplit(value)
    if parts.username or parts.password or parts.path not in ('', '/') or parts.query or parts.fragment or not parts.hostname:
        raise RunnerError('Usa solo indirizzo e porta del Hub, senza percorso o credenziali.')
    if parts.scheme == 'https':
        return value
    try:
        ip = ipaddress.ip_address(parts.hostname)
        private = ip.is_loopback or ip in ipaddress.ip_network('100.64.0.0/10')
    except ValueError:
        private = False
    if parts.scheme != 'http' or not private:
        raise RunnerError('HTTP ammesso solo su loopback o IP Tailscale; altrimenti usa HTTPS.')
    return value


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        return None


class Client:
    def __init__(self, url):
        self.url = validate_url(url)
        self.token = None
        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())

    def request(self, path, payload=None):
        headers = {'Accept': 'application/json'}
        if self.token:
            headers['Authorization'] = 'Bearer ' + self.token
        body = None
        if payload is not None:
            body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
            headers['Content-Type'] = 'application/json'
        req = urllib.request.Request(self.url + path, data=body, headers=headers)
        try:
            with self.opener.open(req, timeout=5) as response:
                raw = response.read(900001)
                if len(raw) > 900000:
                    raise RunnerError('Risposta Hub troppo grande.')
                return json.loads(raw)
        except urllib.error.HTTPError as error:
            if error.code == 401:
                if path == '/runner/pair':
                    raise RunnerError('Codice scaduto, già usato o non valido per questo Hub. Crea una nuova sessione.') from None
                raise RunnerError('Sessione chiusa, scaduta o invalidata dal riavvio del Hub.') from None
            if error.code == 429:
                raise RunnerError('Troppi tentativi. Attendi un minuto e crea una nuova sessione.') from None
            if error.code == 422 and path == '/runner/pair':
                raise RunnerError('Il Hub non accetta Linux. Aggiornalo alla v0.3.') from None
            if error.code == 404 and path == '/runner/health':
                raise RunnerError('Il Hub non dispone del controllo compatibilità Linux. Aggiornalo alla v0.3.') from None
            raise RunnerError(f'Richiesta rifiutata dal Hub: HTTP {error.code}.') from None
        except (urllib.error.URLError, TimeoutError, OSError) as error:
            raise RunnerError('Hub non raggiungibile: ' + str(getattr(error, 'reason', error))) from None
        except (json.JSONDecodeError, UnicodeDecodeError):
            raise RunnerError('Il Hub non ha restituito una risposta JSON valida.') from None


class Guardian:
    """Starts once before threads; kills registered process groups on parent pipe EOF."""
    def __init__(self):
        if not sys.platform.startswith('linux') or threading.active_count() != 1:
            raise RunnerError('Avvia il runner su Linux, prima di altri thread.')
        read_fd, write_fd = os.pipe()  # Non-inheritable (close-on-exec).
        self.pid = os.fork()
        if self.pid == 0:
            os.close(write_fd)
            # Terminal interruptions belong to the interactive parent. Watch its pipe instead.
            for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
                signal.signal(sig, signal.SIG_IGN)
            groups = set()
            remainder = b''
            try:
                while True:
                    data = os.read(read_fd, 4096)
                    if not data:
                        break
                    remainder += data
                    while b'\n' in remainder:
                        line, remainder = remainder.split(b'\n', 1)
                        action, number = line.split(b' ', 1)
                        pgid = int(number)
                        if action == b'ADD':
                            groups.add(pgid)
                        elif action == b'REMOVE':
                            groups.discard(pgid)
            finally:
                for pgid in groups:
                    kill_group(pgid)
                os.close(read_fd)
                os._exit(0)
        os.close(read_fd)
        self.write_fd = write_fd

    def add(self, pgid):
        os.write(self.write_fd, f'ADD {pgid}\n'.encode())

    def remove(self, pgid):
        os.write(self.write_fd, f'REMOVE {pgid}\n'.encode())

    def close(self):
        if self.write_fd is not None:
            os.close(self.write_fd)
            self.write_fd = None
            os.waitpid(self.pid, 0)


def kill_group(pgid):
    try:
        os.killpg(pgid, signal.SIGKILL)
    except ProcessLookupError:
        pass


def validate_command(command):
    if not isinstance(command, dict) or command.get('shell') not in ('bash', 'sh'):
        raise RunnerError('Il Hub ha inviato una shell non supportata su Linux.')
    script = command.get('script')
    timeout = command.get('timeout_seconds')
    if not isinstance(script, str) or not 1 <= len(script) <= 8000 or '\0' in script:
        raise RunnerError('Script non valido.')
    if type(timeout) is not int or not 1 <= timeout <= 120:
        raise RunnerError('Timeout non valido.')
    cid = command.get('id')
    if not isinstance(cid, str) or not cid or len(cid) > 128:
        raise RunnerError('Identificatore comando non valido.')


def run_command(command, guardian, stop):
    validate_command(command)
    shell = '/bin/bash' if command['shell'] == 'bash' else '/bin/sh'
    if not os.path.exists(shell):
        return {'stdout': '', 'stderr': f'Shell non disponibile: {shell}', 'exit_code': -1,
                'duration_ms': 0, 'outcome': 'failed', 'truncated': False}
    # The fixed gate waits. The approved script cannot start before Guardian registration.
    if command['shell'] == 'bash':
        gate = 'IFS= read -r gate || exit 125; [ "$gate" = GO ] || exit 125; exec /bin/bash --noprofile --norc -c "$1"'
        args = [shell, '--noprofile', '--norc', '-c', gate, 'support-runner-gate', command['script']]
    else:
        gate = 'IFS= read -r gate || exit 125; [ "$gate" = GO ] || exit 125; exec /bin/sh -c "$1"'
        args = [shell, '-c', gate, 'support-runner-gate', command['script']]
    started = time.monotonic()
    env = os.environ.copy()
    # Shell startup hooks must not bypass approval or create hidden execution.
    for key in ('BASH_ENV', 'ENV', 'SHELLOPTS', 'BASHOPTS', 'CDPATH', 'GLOBIGNORE'):
        env.pop(key, None)
    env = {key: value for key, value in env.items() if not key.startswith('BASH_FUNC_')}
    env['LC_ALL'] = 'C.UTF-8'
    process = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               start_new_session=True, close_fds=True, cwd=Path.home(), env=env)
    registered = False
    parts = {'stdout': [], 'stderr': []}
    sizes = {'stdout': 0, 'stderr': 0}
    decoders = {name: codecs.getincrementaldecoder('utf-8')('replace') for name in parts}
    truncated = False
    outcome = 'completed'
    exit_code = -1

    def collect(name, data, final=False):
        nonlocal truncated
        text = decoders[name].decode(data, final=final)
        keep = min(len(text), OUTPUT_LIMIT - sizes[name])
        if keep:
            parts[name].append(text[:keep]); sizes[name] += keep
        truncated |= keep < len(text)

    try:
        guardian.add(process.pid)
        registered = True
        if stop.is_set():
            raise RunnerError('Sessione fermata prima dell’esecuzione.')
        process.stdin.write(b'GO\n'); process.stdin.close()
        deadline = started + command['timeout_seconds']
        cleanup_deadline = None
        with selectors.DefaultSelector() as selector:
            for name, stream in [('stdout', process.stdout), ('stderr', process.stderr)]:
                os.set_blocking(stream.fileno(), False)
                selector.register(stream, selectors.EVENT_READ, name)
            while True:
                if stop.is_set():
                    raise RunnerError('Sessione fermata durante l’esecuzione.')
                now = time.monotonic()
                if process.poll() is None and now >= deadline:
                    outcome = 'timed_out'; kill_group(process.pid)
                if process.poll() is not None and cleanup_deadline is None:
                    # Also terminate descendants keeping stdout/stderr open after shell exit.
                    exit_code = process.returncode if outcome == 'completed' else -1
                    kill_group(process.pid)
                    cleanup_deadline = now + 5
                if not selector.get_map() and process.poll() is not None:
                    break
                if cleanup_deadline is not None and now >= cleanup_deadline:
                    raise RunnerError('Un processo ha mantenuto aperto lo stream oltre il limite.')
                for key, _ in selector.select(0.1):
                    data = os.read(key.fileobj.fileno(), 8192)
                    collect(key.data, data, final=not data)
                    if not data:
                        selector.unregister(key.fileobj)
    finally:
        kill_group(process.pid)
        process.wait(timeout=5)
        for stream in (process.stdin, process.stdout, process.stderr):
            if not stream.closed:
                stream.close()
        if registered:
            guardian.remove(process.pid)
    return {'stdout': ''.join(parts['stdout']), 'stderr': ''.join(parts['stderr']), 'exit_code': exit_code,
            'duration_ms': int((time.monotonic() - started) * 1000), 'outcome': outcome, 'truncated': truncated}


class Heartbeat:
    def __init__(self, client, expires_at, stop):
        self.client, self.expires_at, self.stop = client, expires_at, stop
        self.done = threading.Event()
        self.error = None
        self.thread = threading.Thread(target=self.loop, daemon=True)

    def loop(self):
        try:
            while not self.done.wait(2):
                if self.stop.is_set():
                    return
                if time.time() >= self.expires_at:
                    raise RunnerError('Sessione scaduta.')
                reply = self.client.request('/runner/poll')
                if reply.get('command') is not None:
                    raise RunnerError('Il Hub ha inviato un secondo comando inatteso.')
        except Exception as error:
            self.error = error; self.stop.set()

    def __enter__(self):
        self.thread.start(); return self

    def __exit__(self, *args):
        self.done.set(); self.thread.join(timeout=6)
        if self.thread.is_alive():
            self.stop.set()
            raise RunnerError('Il controllo di connessione non si è fermato.')
        if self.error:
            raise RunnerError(str(self.error))


def approval(command, stop, expires_at):
    validate_command(command)
    print('\n' + '=' * 60)
    print(f"Shell: {command['shell']} | Timeout: {command['timeout_seconds']} s")
    print('Comando proposto (caratteri di controllo mostrati come escape):')
    print(safe_text(command['script']))
    print('-' * 60)
    print('Approvi? [s = Approve / Invio = Reject / q = Disconnect] ', end='', flush=True)
    with selectors.DefaultSelector() as selector:
        selector.register(sys.stdin, selectors.EVENT_READ)
        while not stop.is_set():
            if time.time() >= expires_at:
                stop.set(); break
            if not selector.select(0.2):
                continue
            line = sys.stdin.readline()
            if not line or line.strip().lower() == 'q':
                stop.set(); break
            answer = line.strip().lower()
            if answer in ('s', 'approve'):
                return True
            if answer in ('', 'n', 'reject'):
                return False
            print('Usa s, Invio o q: ', end='', flush=True)
    raise RunnerError('Sessione fermata.')


def session_loop(client, guardian, stop, expires_at):
    while not stop.is_set():
        if time.time() >= expires_at:
            raise RunnerError('Sessione scaduta.')
        reply = client.request('/runner/poll')
        command = reply.get('command')
        if command is None:
            stop.wait(1); continue
        validate_command(command)
        with Heartbeat(client, expires_at, stop):
            yes = approval(command, stop, expires_at)
            if stop.is_set():
                raise RunnerError('Sessione fermata.')
            cid = urllib.parse.quote(command['id'], safe='')
            client.request(f'/runner/commands/{cid}/decision', {'decision': 'approve' if yes else 'reject'})
            if not yes:
                print('Comando rifiutato.'); continue
            print('Esecuzione in corso… (Ctrl+C per fermare la sessione)', flush=True)
            result = run_command(command, guardian, stop)
            if stop.is_set():
                raise RunnerError('Connessione persa: risultato non inviato.')
            client.request(f'/runner/commands/{cid}/result', result)
            print(f"Risultato inviato: {result['outcome']}, exit {result['exit_code']}, {result['duration_ms']} ms.")
        print('In attesa del prossimo comando. Ctrl+C per disconnettere.', flush=True)


def main():
    parser = argparse.ArgumentParser(description='SupportRunner Linux temporaneo, con approvazione obbligatoria.')
    parser.add_argument('--hub', default=DEFAULT_HUB)
    parser.add_argument('--version', action='version', version='SupportRunner Linux ' + VERSION)
    args = parser.parse_args()
    if not sys.platform.startswith('linux'):
        parser.error('Questo runner richiede Linux.')
    if not sys.stdin.isatty() or not sys.stdout.isatty():
        parser.error('Apri un terminale interattivo. Avvio unattended o input da pipe non consentiti.')
    client = Client(args.hub)
    print('SupportRunner Linux v' + VERSION)
    print('Hub: ' + safe_text(client.url))
    health = client.request('/runner/health')
    if 'linux' not in health.get('runner_platforms', []):
        raise RunnerError('Aggiorna il Hub alla v0.3 per collegare Linux.')
    code = normalize_code(input('Codice sessione dal pannello: '))
    if not code:
        raise RunnerError('Codice non inserito.')
    stop = threading.Event()
    with_guardian = Guardian()
    paired = False
    try:
        reply = client.request('/runner/pair', {'pairing_token': code, 'machine': socket.gethostname()[:100], 'platform': 'linux'})
        code = ''
        client.token = reply['runner_token']; paired = True
        expires_at = float(reply['expires_at'])
        for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
            signal.signal(sig, lambda *_: stop.set())
        print('Connesso. Privilegi dell’utente corrente; nessuna elevazione automatica.')
        print('In attesa di comandi. Ctrl+C per disconnettere.', flush=True)
        session_loop(client, with_guardian, stop, expires_at)
        return 0
    finally:
        stop.set()
        # Local processes stop before network revocation (best effort, max 5 seconds).
        with_guardian.close()
        if paired:
            try:
                client.request('/runner/disconnect', {})
            except RunnerError:
                pass
        client.token = None
        print('\nDisconnesso.')


if __name__ == '__main__':
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print('\nInterrotto.'); sys.exit(130)
    except (RunnerError, OSError) as error:
        print('\nSessione fermata: ' + safe_text(error), file=sys.stderr); sys.exit(1)