#!/usr/bin/env python3
# ==========================================================
# ProxMenux - A menu-driven script for Proxmox VE management
# Author      : VAIO73
# GitHub      : github.com/Vaso73
# Maintainer  : MacRimi
# Copyright   : (c) 2026 MacRimi & contributors
# License     : (GPL-3.0) (https://github.com/MacRimi/ProxMenux/blob/main/LICENSE)
# Version     : 1.0
# Last Updated: 14/10/2026
# ==========================================================
# Description:
# On-demand native LXC discovery from service and installation metadata.
# Reads bounded data only; never executes discovered programs, starts guests,
# registers applications or infers their update mechanism.
# ==========================================================
"""Separate from catalog detection: callers must explicitly request a scan."""

import copy
import fcntl
import hashlib
import json
import os
from pathlib import Path
import re
import selectors
import signal
import subprocess
import threading
import time

TOTAL_SECONDS = 20
PROBE_SECONDS = 3
OUTPUT_BYTES = 64 * 1024
MAX_RECORDS = 256
MAX_CANDIDATES = 50
CACHE_SECONDS = 300
LOCAL_CONFIG = Path('/etc/pve/local/lxc')
OCI_ROOT = Path('/usr/local/share/proxmenux/oci/instances')
UPDATE_LOCK_ROOT = Path('/run/lock')
_cache = {}
_locks = {}
_mutex = threading.Lock()
_PATH = re.compile(r'/[A-Za-z0-9_.+/@-]{1,510}\Z')
_NAME = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.+@-]{0,63}\Z')
_VERSION = re.compile(r'v?\d+(?:\.\d+){1,4}(?:[-+._][A-Za-z0-9.-]+)?\Z')
_SYSTEM = {
    'ssh', 'sshd', 'cron', 'crond', 'postfix', 'rsyslog', 'syslog', 'syslog-ng',
    'dbus', 'dbus-broker', 'systemd', 'udev', 'networking', 'network', 'networkmanager',
    'dhcpcd', 'dhclient', 'chrony', 'chronyd', 'ntp', 'ntpd', 'openntpd', 'agetty',
    'getty', 'login', 'polkit', 'polkitd', 'accounts-daemon', 'acpid', 'anacron',
    'docker', 'containerd', 'containerd-shim', 'podman', 'qemu-guest-agent', 'rngd',
    'auditd', 'logrotate', 'local', 'bootmisc', 'hostname', 'hwclock', 'modules', 'master',
    'proxmox-boot-sync', 'pve-container', 'pve-firewall', 'pvefw-logger',
    'pvedaemon', 'pvestatd', 'pveproxy', 'pvescheduler', 'spiceproxy', 'lxc',
    'mount-ro', 'mdev', 'sysctl', 'sysfs', 'devfs', 'cgroups', 'rc', 'rcs', 'swap',
    'sleep', 'true', 'false', 'echo', 'cat', 'head', 'tail', 'tee', 'logger',
    'apt', 'apt-get', 'dpkg', 'dpkg-query', 'apk', 'unattended-upgrades',
    'fstrim', 'fsck', 'fsckd', 'e2scrub', 'e2scrub-all', 'e2scrub-reap',
    'irqbalance', 'smartd', 'smartmontools', 'rpcbind', 'nfs-common', 'nfs-server',
    'rpc-statd', 'rpc-statd-notify', 'rpc-gssd', 'rpc-svcgssd', 'nftables', 'iptables',
    'ip6tables', 'firewalld', 'ufw', 'resolvconf', 'wpa-supplicant', 'modemmanager',
    'console-setup', 'keyboard-setup', 'plymouth', 'halt', 'reboot', 'shutdown',
    'rc-local', 'ntpsec', 'networkd-dispatcher', 'busybox', 'atd', 'uuidd',
}
_RUNTIMES = re.compile(r'(?:python[0-9.]*|node(?:js)?|sh|bash|dash|env|java|ruby|perl)\Z')


class ProbeError(Exception):
    """Safe error code only; guest output must never become an API error."""


def bounded_run(argv, deadline):
    """Cap combined stdout/stderr while reading, and kill/reap the process group."""
    remaining = min(PROBE_SECONDS, deadline - time.monotonic())
    if remaining <= 0:
        raise ProbeError('deadline')
    end = time.monotonic() + remaining
    try:
        proc = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                start_new_session=True)
    except OSError:
        raise ProbeError('unavailable') from None
    outputs = [bytearray(), bytearray()]
    try:
        with selectors.DefaultSelector() as poll:
            for index, stream in enumerate((proc.stdout, proc.stderr)):
                os.set_blocking(stream.fileno(), False)
                poll.register(stream, selectors.EVENT_READ, index)
            while poll.get_map():
                wait = end - time.monotonic()
                if wait <= 0:
                    raise ProbeError('timeout')
                for key, _ in poll.select(wait):
                    block = os.read(key.fileobj.fileno(), 4096)
                    if not block:
                        poll.unregister(key.fileobj)
                        continue
                    if sum(map(len, outputs)) + len(block) > OUTPUT_BYTES:
                        raise ProbeError('output_limit')
                    outputs[key.data].extend(block)
            try:
                proc.wait(timeout=max(0.001, end - time.monotonic()))
            except subprocess.TimeoutExpired:
                raise ProbeError('timeout') from None
        if proc.returncode:
            raise ProbeError('read_failed')
        return outputs[0].decode('utf-8', errors='strict')
    except UnicodeError:
        raise ProbeError('invalid_data') from None
    finally:
        # Also clean up descendants that outlive a successfully exited wrapper.
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        proc.wait()
        proc.stdout.close()
        proc.stderr.close()


# These are fixed read-only scripts, not guest-supplied shell or init scripts.
# awk emits only executable/script paths and WorkingDirectory, never arguments
# or environment values. OpenRC files are parsed as data, never sourced.
SERVICE_SCRIPT = r'''
command -v awk >/dev/null && command -v head >/dev/null || exit 1
n=0
for dir in /etc/systemd/system /usr/lib/systemd/system /lib/systemd/system /etc/init.d; do
  [ ! -d "$dir" ] || [ -r "$dir" ] || exit 1
  for f in "$dir"/*; do
    [ -f "$f" ] || continue
    case "$dir:$f" in /etc/init.d:*) kind=openrc;; *:*.service) kind=systemd;; *) continue;; esac
    n=$((n+1)); [ "$n" -le 256 ] || { printf 'LIMIT\n'; exit 0; }
    [ -r "$f" ] || exit 1
    content=$(timeout -s KILL 1 head -c 16385 "$f") || exit 1
    [ "${#content}" -le 16384 ] || { printf 'LIMIT\n'; exit 0; }
    printf '%s\n' "$content" | awk -v kind="$kind" -v file="$f" '
      function clean(v) {gsub(/^[ \t]+|[ \t]+$/, "", v); gsub(/^["\047]|["\047]$/, "", v); return v}
      /^ExecStart=/ {v=substr($0,11); sub(/^[-+!:@]+/, "", v); split(v,a,/[ \t]+/); exe=clean(a[1]); script=clean(a[2]); if(script !~ /^\//) script=""}
      /^WorkingDirectory=/ {dir=clean(substr($0,18))}
      /^command=/ {exe=clean(substr($0,9))}
      /^directory=/ {dir=clean(substr($0,11))}
      /^command_args=/ {v=clean(substr($0,14)); split(v,a,/[ \t]+/); script=clean(a[1]); if(script !~ /^\//) script=""}
      END {printf "S\t%s\t%s\t%s\t%s\t%s\n", file, kind, exe, script, dir}'
  done
done
printf 'END\n'
'''
PROCESS_SCRIPT = r'''
command -v readlink >/dev/null && command -v head >/dev/null || exit 1
n=0
for dir in /proc/[0-9]*; do
  [ -d "$dir" ] || continue
  n=$((n+1)); [ "$n" -le "$1" ] || { printf 'LIMIT\n'; exit 0; }
  # Never inspect Docker/OCI child workloads or read cmdline/environ.
  [ -d "$dir" ] || continue
  [ -r "$dir/cgroup" ] || exit 1
  group=$(timeout -s KILL 1 head -c 8193 "$dir/cgroup") || exit 1
  [ "${#group}" -le 8192 ] || { printf 'LIMIT\n'; exit 0; }
  case "$group" in *docker*|*libpod*|*kubepods*|*containerd*) continue;; esac
  # Kernel threads and vanished processes have no application executable.
  [ -L "$dir/exe" ] || continue
  exe=$(timeout -s KILL 1 readlink "$dir/exe") || continue
  cwd=$(timeout -s KILL 1 readlink "$dir/cwd") || continue
  printf 'P\t%s\t%s\t%s\n' "${dir##*/}" "$exe" "$cwd"
done
printf 'END\n'
'''
PORT_SCRIPT = r'''
command -v awk >/dev/null || exit 1
if command -v ss >/dev/null; then
  (timeout -s KILL 1 ss -H -lntp || printf 'FAILED\n') | awk -v cap="$1" '
    /^FAILED$/ {failed=1; exit}
    NR > cap {limited=1; exit}
    {port=$4; sub(/^.*:/,"",port); rest=$0
     # More than one owner is ambiguous; never choose the first arbitrarily.
     if (gsub(/pid=/,"pid=",rest)!=1) next
     if (match(rest,/pid=[0-9]+/)) {pid=substr(rest,RSTART+4,RLENGTH-4); printf "L\t%s\t%s\n",pid,port}}
    END {print failed ? "FAILED" : (limited ? "LIMIT" : "END")}'
elif command -v netstat >/dev/null; then
  (timeout -s KILL 1 netstat -lntp || printf 'FAILED\n') | awk -v cap="$1" '
    /^FAILED$/ {failed=1; exit}
    /^tcp/ {n++; if(n>cap) {limited=1; exit}; port=$4; sub(/^.*:/,"",port)
      pid=$7; sub(/\/.*$/,"",pid); if(pid ~ /^[0-9]+$/) printf "L\t%s\t%s\n",pid,port}
    END {print failed ? "FAILED" : (limited ? "LIMIT" : "END")}'
else
  exit 1
fi
'''
FOOTPRINT_SCRIPT = r'''
[ -f "$1" ] && [ -r "$1" ] || exit 1
printf 'FILE\n'
if [ -f "$2/package.json" ]; then
  [ -r "$2/package.json" ] || exit 1
  timeout -s KILL 1 head -c 8192 "$2/package.json"
fi
'''


def _path(value):
    return bool(isinstance(value, str) and _PATH.fullmatch(value)
                and '..' not in value.split('/'))


def _system(name):
    raw = name.lower().removesuffix('.service')
    stem = raw.split('@')[0]
    return raw.startswith('user@') or stem in _SYSTEM or stem.startswith((
        'systemd-', 'dbus-', 'getty', 'containerd-', 'apt-', 'dpkg-', 'e2scrub',
        'unattended-', 'ifup', 'ifdown', 'user@', 'btrfs-', 'lvm2-', 'lxc-'))


def _custom(path):
    return _path(path) and path.startswith(('/opt/', '/srv/', '/usr/local/'))


def parse_records(text, kind):
    lines = text.splitlines()
    if not lines or lines[-1] not in ('END', 'LIMIT'):
        raise ProbeError('incomplete')
    if len(lines) - 1 > MAX_RECORDS:
        raise ProbeError('metadata_limit')
    records = []
    seen_services = set()
    for line in lines[:-1]:
        parts = line.split('\t')
        if kind == 'S' and len(parts) == 6 and parts[0] == 'S':
            _, source, manager, exe, script, directory = parts
            if not _path(source) or manager not in ('systemd', 'openrc'):
                raise ProbeError('invalid_data')
            service = Path(source).name.removesuffix('.service')
            if service in seen_services:
                continue  # /etc overrides vendor units; no duplicate installations.
            seen_services.add(service)
            if not _NAME.fullmatch(service) or _system(service) or not _path(exe):
                continue
            runtime = bool(_RUNTIMES.fullmatch(Path(exe).name))
            install = script if runtime else exe
            if (not _path(install) or _system(Path(install).name)
                    or install.startswith(('/usr/lib/systemd/', '/lib/systemd/',
                                           '/usr/lib/udev/', '/lib/udev/', '/usr/lib/apt/'))):
                continue
            root = directory if _custom(directory) else str(Path(install).parent)
            # Generic runtime without an absolute application script is ambiguous.
            if runtime and not _custom(install):
                continue
            records.append({'name': service, 'path': install, 'root': root,
                            'service': source, 'manager': manager, 'executable': exe})
        elif kind == 'P' and len(parts) == 4 and parts[0] == 'P':
            _, pid, exe, directory = parts
            if not pid.isdigit() or not _path(exe) or not _path(directory):
                continue
            records.append({'pid': pid, 'executable': exe, 'root': directory})
        elif kind == 'L' and len(parts) == 3 and parts[0] == 'L':
            _, pid, port = parts
            if pid.isdigit() and port.isdigit() and 0 < int(port) < 65536:
                records.append({'pid': pid, 'port': int(port)})
            else:
                raise ProbeError('invalid_data')
        else:
            raise ProbeError('invalid_data')
    return records, lines[-1] == 'LIMIT', len(lines) - 1


def candidate(vmid, record, footprint):
    if not footprint.startswith('FILE\n'):
        raise ProbeError('invalid_data')
    version, metadata_issue = None, None
    metadata = footprint[5:].strip()
    if metadata:
        try:
            manifest = json.loads(metadata)
        except (ValueError, TypeError):
            manifest = None
        # Only the whitelisted version field leaves this function. Never retain
        # manifest scripts, repository guesses, config or arbitrary extra fields.
        if not isinstance(manifest, dict):
            metadata_issue = 'invalid_metadata'
            manifest = {}
        value = manifest.get('version')
        if (isinstance(value, str) and len(value) <= 64 and _VERSION.fullmatch(value)
                and record['path'].startswith(record['root'].rstrip('/') + '/')):
            version = value
    identity = 'native-' + hashlib.sha256(
        f'{int(vmid)}:{record["path"]}:{record["root"]}'.encode()).hexdigest()[:24]
    evidence = [{'kind': 'installation_file', 'value': record['path']}]
    if record.get('service'):
        evidence.append({'kind': record['manager'], 'value': record['service']})
    else:
        evidence.extend([{'kind': 'process_executable', 'value': record['process']},
                         {'kind': 'working_directory', 'value': record['root']}])
    ports = sorted(set(record.get('listening_ports') or []))
    evidence.extend({'kind': 'tcp_listener', 'value': str(port)} for port in ports)
    return {'slug': identity, 'discovery_id': identity, 'source': 'native_metadata',
            'name': record['name'], 'installed_version': version, 'evidence': evidence,
            'installation_path': record['path'], 'default_ports': [],
            'listening_ports': ports,
            'metadata_issue': metadata_issue,
            'tracking_suggestion': ({'installed_via': 'manual', 'installed_version': version,
                                     'detected_version': version, 'detector_verified': True}
                                    if version else None)}


def _guard(vmid, deadline, run):
    # The local node path rejects QEMU, missing and remote CTs before pct exec.
    path = LOCAL_CONFIG / f'{vmid}.conf'
    if not path.is_file():
        raise ProbeError('not_local_lxc')
    # Presence, including a damaged/in-progress record, excludes generic probes.
    try:
        os.lstat(OCI_ROOT / str(vmid))
    except FileNotFoundError:
        pass
    except OSError:
        raise ProbeError('oci_unknown') from None
    else:
        raise ProbeError('managed_oci')
    try:
        with path.open('rb') as handle:
            raw = handle.read(OUTPUT_BYTES + 1)
        if len(raw) > OUTPUT_BYTES:
            raise ProbeError('config_limit')
        config = raw.decode('utf-8')
    except (OSError, UnicodeError):
        raise ProbeError('config_unavailable') from None
    if re.search(r'^lock:', config, re.M):
        raise ProbeError('guest_busy')
    update_lock = UPDATE_LOCK_ROOT / f'proxmenux-lxc-update-{vmid}.lock'
    try:
        with update_lock.open('rb') as handle:
            try:
                fcntl.flock(handle, fcntl.LOCK_SH | fcntl.LOCK_NB)
            except BlockingIOError:
                raise ProbeError('guest_busy') from None
            finally:
                fcntl.flock(handle, fcntl.LOCK_UN)
    except FileNotFoundError:
        pass
    except OSError:
        raise ProbeError('lock_unavailable') from None
    if re.search(r'^ostype:\s*unmanaged\s*$', config, re.M):
        raise ProbeError('unsupported_guest')
    if not re.search(r'^ostype:\s*(?:debian|ubuntu|alpine|archlinux|centos|fedora|opensuse|gentoo)\s*$', config, re.M):
        raise ProbeError('unsupported_guest')
    status = run(['pct', 'status', str(vmid)], deadline).strip()
    if status != 'status: running':
        raise ProbeError('guest_stopped' if status == 'status: stopped' else 'status_unknown')


def scan(vmid, run=bounded_run):
    deadline = time.monotonic() + TOTAL_SECONDS
    _guard(vmid, deadline, run)
    records, issues, consumed, processes, listeners = [], [], 0, [], []
    for kind, script in (('S', SERVICE_SCRIPT), ('P', PROCESS_SCRIPT), ('L', PORT_SCRIPT)):
        try:
            if consumed >= MAX_RECORDS:
                issues.append('metadata_limit')
                break
            parsed, limited, count = parse_records(run(
                ['pct', 'exec', str(vmid), '--', 'timeout', '-s', 'KILL', '2',
                 'sh', '-c', script, 'discovery', str(MAX_RECORDS - consumed)], deadline), kind)
            if consumed + count > MAX_RECORDS:
                issues.append('metadata_limit')
                parsed = parsed[:MAX_RECORDS - consumed]
            consumed = min(MAX_RECORDS, consumed + count)
            # The entire collector, not each metadata source, has 256 records.
            capacity = MAX_RECORDS - len(records)
            if kind == 'S':
                records.extend(parsed[:capacity])
            elif kind == 'P':
                processes = parsed[:capacity]
            else:
                listeners = parsed[:capacity]
            if limited or len(parsed) > capacity:
                issues.append('metadata_limit')
        except ProbeError as exc:
            issues.append(str(exc))
    for process in processes:
        exe, directory = process['executable'], process['root']
        related = [record for record in records if record['executable'] == exe and (
            not _RUNTIMES.fullmatch(Path(exe).name) or record['root'] == directory)]
        ports = [entry['port'] for entry in listeners if entry['pid'] == process['pid']]
        if len(related) == 1:
            related[0].setdefault('listening_ports', []).extend(ports)
        elif not related and (_custom(exe) and _custom(directory)
                              and not _system(Path(exe).name)
                              and not _RUNTIMES.fullmatch(Path(exe).name)
                              and exe.startswith(directory.rstrip('/') + '/')
                              and _NAME.fullmatch(Path(exe).name)):
            records.append({'name': Path(exe).name, 'path': exe, 'root': directory,
                            'process': exe, 'executable': exe, 'listening_ports': ports})
    items, seen = [], set()
    for record in records:
        key = (record['path'], record['root'])
        if key in seen:
            continue
        seen.add(key)
        if len(items) >= MAX_CANDIDATES:
            issues.append('candidate_limit')
            break
        try:
            footprint = run(['pct', 'exec', str(vmid), '--', 'timeout', '-s', 'KILL', '2',
                             'sh', '-c', FOOTPRINT_SCRIPT,
                             'discovery', record['path'], record['root']], deadline)
            item = candidate(vmid, record, footprint)
            items.append(item)
            if item['metadata_issue']:
                issues.append(item['metadata_issue'])
        except ProbeError as exc:
            issues.append(str(exc))
            if str(exc) == 'deadline':
                break
    return {'vmid': vmid, 'status': 'partial' if issues else 'complete',
            'candidates': items, 'issues': sorted(set(issues)), 'stale': False,
            'scanned_at': time.time()}


def cached(vmid):
    with _mutex:
        value = copy.deepcopy(_cache.get(int(vmid)))
    if value is None:
        return {'vmid': int(vmid), 'status': 'not_scanned', 'candidates': [], 'stale': False}
    if time.time() - value['scanned_at'] >= CACHE_SECONDS:
        value['stale'] = True
    return value


def mark_stale(vmid):
    """Lifecycle events retire freshness, never collect new guest metadata."""
    with _mutex:
        if int(vmid) in _cache:
            _cache[int(vmid)]['stale'] = True


def discover(vmid):
    vmid = int(vmid)
    with _mutex:
        lock = _locks.setdefault(vmid, threading.Lock())
    if not lock.acquire(blocking=False):
        return {**cached(vmid), 'status': 'busy', 'stale': True}
    try:
        try:
            result = scan(vmid)
        except ProbeError as exc:
            result = {**cached(vmid), 'status': 'error', 'issues': [str(exc)], 'stale': True}
            result.setdefault('scanned_at', time.time())
            with _mutex:
                _cache[vmid] = copy.deepcopy(result)
            return result
        previous = cached(vmid)
        if result['status'] != 'complete' and previous['candidates']:
            # Partial output cannot prove that previously seen apps disappeared.
            ids = {item['discovery_id'] for item in result['candidates']}
            result['candidates'].extend(item for item in previous['candidates']
                                        if item['discovery_id'] not in ids)
            result['candidates'] = result['candidates'][:MAX_CANDIDATES]
            result['stale'] = True
        with _mutex:
            _cache[vmid] = copy.deepcopy(result)
        return result
    finally:
        lock.release()


def filter_candidates(result, suggestions, apps):
    """Pure composition: never run catalog probes, load or write registries."""
    known = [*apps, *(suggestions.get('extras') or []),
             *(suggestions.get('docker_workloads') or [])]
    if suggestions.get('helper_slug'):
        known.append({'name': suggestions.get('name_suggestion'),
                      'helper_slug': suggestions['helper_slug'],
                      'tracking_suggestion': suggestions.get('tracking_suggestion')})
    identities = {a.get('discovery_id') for a in known if a.get('discovery_id')}
    names, paths = set(), set()
    for app in known:
        for field in ('name', 'slug', 'helper_slug', 'package'):
            if app.get(field):
                names.add(str(app[field]).casefold())
        detector = app.get('tracking_suggestion') or app
        for field in ('binary_path', 'file_path'):
            if detector.get(field):
                paths.add(detector[field])
    items = [item for item in result['candidates']
             if item['discovery_id'] not in identities
             and item['name'].casefold() not in names
             and item['installation_path'] not in paths]
    return {**result, 'candidates': items}