#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
tor_manager.py
--------------

Gestore delle richieste HTTP/HTTPS attraverso il proxy SOCKS5 di Tor.

Strategia per piattaforma:
  - Windows        → avvia tor.exe bundled (subprocess, Job Object)
  - Linux desktop  → avvia binario tor bundled (subprocess)
  - Android        → copia il binario nella storage INTERNA di Kodi
                     (non-noexec), chmod +x, avvia con subprocess.
                     Fallback: si connette al proxy Orbot se disponibile.
                     Secondo fallback: connessione diretta.

Perché funziona su Android senza root:
  La cartella /data/data/org.xbmc.kodi/ è scrivibile dall'app e
  la sua partizione NON è montata noexec, a differenza della SD card
  e delle cartelle addon su storage esterna.
  Android blocca solo i binari in posizioni non controllate dall'app;
  i binari nella data-dir dell'app stessa sono eseguibili.

Compatibile con Python 3.8 (Kodi 21) → nessun operatore «|» per Union.
"""

import os
import shutil
import stat
import socket
import subprocess
import threading
import time
import logging
import platform
from contextlib import contextmanager, nullcontext
from typing import Any, Dict, List, Optional, Union

import socks          # script.module.pysocks (mantenuto per compatibilita' legacy)
import requests

# ----------------------------------------------------------------------
# Configurazione porta
# ----------------------------------------------------------------------
TOR_HOST       = "127.0.0.1"
TOR_SOCKS_PORT = 9050
TOR_CTRL_PORT  = 9051
TOR_PROXY      = f"socks5h://{TOR_HOST}:{TOR_SOCKS_PORT}"

# ----------------------------------------------------------------------
# APK TorProxy esterno (fallback Android quando il binario bundled fallisce)
# Package/service/activity ricavati dal manifest dell'utente.
# ----------------------------------------------------------------------
TORPROXY_APK_PACKAGE  = "com.torproxy"
TORPROXY_APK_ACTIVITY = "com.torproxy.MainActivity"          # nome completo
TORPROXY_APK_SERVICE  = "com.torproxy.service.OrbotService"  # exported=true nel manifest
TORPROXY_APK_WAIT     = 60                                   # secondi di attesa porta

DEFAULT_HEADERS: Dict[str, str] = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0 Safari/537.36"
    ),
    "Accept": "*/*",
    "Connection": "keep-alive",
}

# ----------------------------------------------------------------------
# Logging
# ----------------------------------------------------------------------
log = logging.getLogger("tor_manager")
if not log.handlers:
    h = logging.StreamHandler()
    h.setFormatter(logging.Formatter(
        "%(asctime)s %(levelname)s [tor_manager] %(message)s"
    ))
    log.addHandler(h)
log.setLevel(logging.DEBUG)


# ======================================================================
# Rilevamento piattaforma
# ======================================================================

def _is_android() -> bool:
    return (
        "ANDROID_ROOT" in os.environ
        or "ANDROID_DATA" in os.environ
        or os.path.isfile("/system/build.prop")
    )


def _is_windows() -> bool:
    return platform.system().lower() == "windows"


# ======================================================================
# GESTIONE PROXY — SENZA MONKEYPATCHING GLOBALE DEL SOCKET
#
# Il vecchio approccio (socket.socket = socks.socksocket) modificava
# il socket a livello globale, bloccando bypass.py e qualsiasi altra
# libreria che usasse requests/socket in quel momento.
#
# La nuova strategia usa ESCLUSIVAMENTE session.proxies di requests,
# che è thread-safe e non interferisce con altri moduli.
# ======================================================================

def _apply_proxy() -> None:
    """Noop — il proxy è gestito a livello di Session, non di socket globale."""
    log.debug("Proxy Tor attivo sulla sessione TorManager")


def _reset_proxy() -> None:
    """Noop — nessun socket globale da ripristinare."""
    log.debug("Proxy Tor: nessun reset socket necessario")


@contextmanager
def tor_enabled():
    """
    Context manager mantenuto per compatibilità con tor_mac_check.py.
    Non altera più il socket globale: il proxy è già configurato sulla
    Session dell'istanza TorManager singleton — non serve fare nulla qui.
    """
    yield


# ======================================================================
# Helper porta aperta
# ======================================================================

def _port_open(host: str, port: int, timeout: float = 2.0) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


# ======================================================================
# Cache noexec + ricerca percorso eseguibile su Android
# ======================================================================

# Cache noexec in memoria (per la sessione corrente)
_noexec_cache: Dict[str, bool] = {}

# Flag: True se su questo dispositivo Android tutti i percorsi sono noexec.
# Viene impostato da _find_executable_dir() dopo il primo giro completo.
_android_confirmed_all_noexec: bool = False


def _android_all_noexec() -> bool:
    """True se abbiamo già verificato che tutti i percorsi Android sono noexec."""
    return _android_confirmed_all_noexec

# Cache noexec persistente su file — sopravvive ai re-import del modulo
_NOEXEC_CACHE_FILE = "/data/local/tmp/_kodi_tor_noexec_cache"


def _load_noexec_cache() -> None:
    """
    Carica la cache noexec dal file persistente.
    Ripristina anche il flag _android_confirmed_all_noexec tra le sessioni
    di Kodi (il flag in memoria si azzera a ogni re-import del modulo).
    """
    global _android_confirmed_all_noexec
    try:
        with open(_NOEXEC_CACHE_FILE, "r") as f:
            for line in f:
                line = line.strip()
                if "=" in line:
                    k, v = line.split("=", 1)
                    if k == "__all_noexec__":
                        _android_confirmed_all_noexec = (v == "1")
                    else:
                        _noexec_cache[k] = (v == "1")
    except Exception:
        pass


def _save_noexec_cache() -> None:
    """Salva la cache noexec su file, incluso il flag __all_noexec__."""
    try:
        with open(_NOEXEC_CACHE_FILE, "w") as f:
            f.write(f"__all_noexec__={'1' if _android_confirmed_all_noexec else '0'}\n")
            for k, v in _noexec_cache.items():
                f.write(f"{k}={'1' if v else '0'}\n")
    except Exception:
        pass


# Carica cache al momento dell'import
_load_noexec_cache()


def _is_noexec(directory: str) -> bool:
    """
    Verifica empiricamente se una directory è su filesystem noexec.
    Risultato cached in memoria e su file per evitare subprocess ripetuti
    anche tra invocazioni diverse dell'addon.
    """
    key = os.path.normpath(directory)
    if key in _noexec_cache:
        log.debug("_is_noexec(%s) = %s (cache)", key, _noexec_cache[key])
        return _noexec_cache[key]

    test_path = os.path.join(directory, "_tor_exec_test")
    result = True
    try:
        with open(test_path, "w") as f:
            f.write("#!/bin/sh\nexit 0\n")
        os.chmod(test_path, 0o755)
        subprocess.call(
            [test_path],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=3,
        )
        result = False   # eseguito → non è noexec
    except (PermissionError, OSError):
        result = True
    except Exception:
        result = True
    finally:
        try:
            os.remove(test_path)
        except Exception:
            pass

    _noexec_cache[key] = result
    _save_noexec_cache()
    log.debug("_is_noexec(%s) = %s", key, result)
    return result


def _find_executable_dir() -> Optional[str]:
    """
    Cerca un percorso scrivibile ed ESEGUIBILE dove copiare il binario Tor.
    Prova in ordine diversi candidati su Android, testando noexec su ognuno.
    Scarta sempre /storage/ e /sdcard/ (sempre noexec).
    """
    pkg = "org.xbmc.kodi"

    def _try(path: str, label: str) -> Optional[str]:
        if not path:
            return None
        if path.startswith("/storage/") or path.startswith("/sdcard/"):
            return None
        try:
            os.makedirs(path, exist_ok=True)
        except OSError:
            return None
        if not os.path.isdir(path):
            return None
        if _is_noexec(path):
            log.debug("_find_exec_dir: noexec (%s): %s", label, path)
            return None
        log.info("_find_exec_dir: OK (%s): %s", label, path)
        return path

    candidates: List[tuple] = []
    for base in ("/data/user/0", "/data/data"):
        candidates += [
            (os.path.join(base, pkg, "cache", "apk", "tor"), "cache/apk"),
            (os.path.join(base, pkg, "files", "tor"),         "files"),
            (os.path.join(base, pkg, "cache", "tor"),         "cache"),
        ]

    # Ricava percorso dal Python di Kodi in esecuzione
    try:
        self_path = os.path.abspath(__file__)
        parts = self_path.split(os.sep)
        if pkg in parts:
            idx = parts.index(pkg)
            kodi_root = os.sep.join(parts[: idx + 1])
            for sub in ("cache/apk/tor", "cache/tor", "files/tor"):
                candidates.append(
                    (os.path.join(kodi_root, sub), "__file__/" + sub)
                )
    except Exception:
        pass

    candidates.append(("/data/local/tmp/kodi_tor", "data/local/tmp"))

    for path, label in candidates:
        result = _try(path, label)
        if result:
            return result

    # Tutti i percorsi sono noexec — imposta il flag globale per evitare
    # di ripetere i test ad ogni invocazione dell'addon
    global _android_confirmed_all_noexec
    _android_confirmed_all_noexec = True
    _save_noexec_cache()   # persiste il flag su disco per le sessioni future
    log.error("_find_exec_dir: tutti i filesystem sono noexec su questo dispositivo")
    return None


def _get_user_proxy() -> Optional[str]:
    """
    Legge le impostazioni proxy dall'addon Kodi.
    Prova sia getSetting (stringa) che getSettingBool per massima compatibilità.
    """
    try:
        import xbmcaddon
        addon = xbmcaddon.Addon()

        # getSetting restituisce sempre stringa con settings.xml sintassi vecchia
        # getSettingBool lancia eccezione con type="bool" (vecchia sintassi)
        # → usiamo SOLO getSetting stringa per massima compatibilità
        raw = addon.getSetting("proxy_enabled").strip().lower()
        enabled = raw in ("true", "1", "yes")
        log.debug("_get_user_proxy: proxy_enabled raw=%r → %s", raw, enabled)

        if not enabled:
            log.debug("_get_user_proxy: proxy disabilitato nelle impostazioni")
            return None

        host = addon.getSetting("proxy_host").strip()
        log.debug("_get_user_proxy: proxy_host=%r", host)
        if not host:
            log.warning("_get_user_proxy: proxy abilitato ma host vuoto")
            return None

        port  = addon.getSetting("proxy_port").strip() or "8080"
        ptype = addon.getSetting("proxy_type").strip()   # "0"=http "1"=socks5
        user  = addon.getSetting("proxy_user").strip()
        pwd   = addon.getSetting("proxy_pass").strip()

        log.debug("_get_user_proxy: port=%s type=%s user=%r", port, ptype, user)

        scheme = "socks5" if ptype == "1" else "http"

        if user and pwd:
            proxy_url = f"{scheme}://{user}:{pwd}@{host}:{port}"
        elif user:
            proxy_url = f"{scheme}://{user}@{host}:{port}"
        else:
            proxy_url = f"{scheme}://{host}:{port}"

        log.info("Proxy utente attivo: %s", proxy_url)
        return proxy_url

    except Exception as e:
        log.error("_get_user_proxy errore: %s", e)
        return None


# ======================================================================
# Avvio APK TorProxy esterno su Android
# ======================================================================

def _check_apk_installed(package: str) -> bool:
    """
    Verifica se un APK è installata tramite 'pm list packages'.
    Restituisce True se il package è presente.
    """
    try:
        result = subprocess.check_output(
            ["pm", "list", "packages", package],
            stderr=subprocess.DEVNULL,
            timeout=5,
        ).decode("utf-8", errors="replace")
        installed = package in result
        log.info("_check_apk_installed(%s): %s", package, installed)
        return installed
    except Exception as e:
        log.warning("_check_apk_installed errore: %s", e)
        return False


def _launch_torproxy_apk_nonblocking() -> bool:
    """
    Avvia l'APK TorProxy usando l'API nativa di Kodi (StartAndroidActivity).

    NON usa 'am' via subprocess: su Android moderno Kodi non ha i permessi
    per avviare servizi/activity di altre app tramite shell (codice 255).
    xbmc.executebuiltin('StartAndroidActivity(...)') bypassa questo limite
    perché è eseguito nel contesto del processo Kodi stesso.

    NON attende la porta — il chiamante deve monitorarla in un thread
    daemon per non bloccare il thread principale di Kodi (che ha un timeout
    di ~5 secondi prima di killare lo script).

    Ritorna True se il comando è stato inviato (NON se Tor è già pronto).
    """
    if not _is_android():
        return False

    # Porta già aperta → APK già attiva
    if _port_open(TOR_HOST, TOR_SOCKS_PORT, timeout=2):
        log.info("TorProxy APK già attivo su %s:%s", TOR_HOST, TOR_SOCKS_PORT)
        return True

    if not _check_apk_installed(TORPROXY_APK_PACKAGE):
        log.error(
            "TorProxy APK '%s' NON installata sul dispositivo. "
            "Installa l'APK e riprova.", TORPROXY_APK_PACKAGE
        )
        try:
            import xbmcgui
            xbmcgui.Dialog().notification(
                "TorProxy",
                f"APK '{TORPROXY_APK_PACKAGE}' non trovata. Installala per usare Tor.",
                xbmcgui.NOTIFICATION_WARNING,
                5000,
            )
        except Exception:
            pass
        return False

    # Usa StartAndroidActivity — API Kodi nativa, funziona senza permessi shell.
    # Sintassi: StartAndroidActivity(package, intent_action, data_type, data_URI)
    # Avviamo la MainActivity che a sua volta avvierà il servizio Tor interno.
    try:
        import xbmc as _xbmc
        builtin = (
            f"StartAndroidActivity("
            f"{TORPROXY_APK_PACKAGE},"   # package
            f","                          # intent action (vuoto = MAIN)
            f","                          # data type
            f""                           # data URI
            f")"
        )
        log.info("TorProxy APK: StartAndroidActivity(%s)", TORPROXY_APK_PACKAGE)
        _xbmc.executebuiltin(builtin)
        return True
    except Exception as exc:
        log.error("StartAndroidActivity fallito: %s", exc)
        return False


# ======================================================================
# Helper Windows: Job Object
# ======================================================================

def _assign_job_object(pid: int) -> Optional[object]:
    try:
        import ctypes
        import ctypes.wintypes as wt

        PROCESS_ALL_ACCESS         = 0x1F0FFF
        JobObjectExtendedLimitInfo = 9
        JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x2000

        class JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure):
            _fields_ = [
                ("PerProcessUserTimeLimit", wt.LARGE_INTEGER),
                ("PerJobUserTimeLimit",     wt.LARGE_INTEGER),
                ("LimitFlags",             wt.DWORD),
                ("MinimumWorkingSetSize",   ctypes.c_size_t),
                ("MaximumWorkingSetSize",   ctypes.c_size_t),
                ("ActiveProcessLimit",      wt.DWORD),
                ("Affinity",               ctypes.POINTER(wt.ULONG)),
                ("PriorityClass",          wt.DWORD),
                ("SchedulingClass",        wt.DWORD),
            ]

        class IO_COUNTERS(ctypes.Structure):
            _fields_ = [(f, wt.ULARGE_INTEGER) for f in (
                "ReadOperationCount", "WriteOperationCount",
                "OtherOperationCount", "ReadTransferCount",
                "WriteTransferCount", "OtherTransferCount",
            )]

        class JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure):
            _fields_ = [
                ("BasicLimitInformation", JOBOBJECT_BASIC_LIMIT_INFORMATION),
                ("IoInfo",                IO_COUNTERS),
                ("ProcessMemoryLimit",    ctypes.c_size_t),
                ("JobMemoryLimit",        ctypes.c_size_t),
                ("PeakProcessMemoryUsed", ctypes.c_size_t),
                ("PeakJobMemoryUsed",     ctypes.c_size_t),
            ]

        k32 = ctypes.windll.kernel32  # type: ignore[attr-defined]
        job = k32.CreateJobObjectW(None, None)
        if not job:
            return None

        info = JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
        info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
        ok = k32.SetInformationJobObject(
            job, JobObjectExtendedLimitInfo,
            ctypes.byref(info), ctypes.sizeof(info),
        )
        if not ok:
            k32.CloseHandle(job)
            return None

        proc_handle = k32.OpenProcess(PROCESS_ALL_ACCESS, False, pid)
        if not proc_handle:
            k32.CloseHandle(job)
            return None

        ok = k32.AssignProcessToJobObject(job, proc_handle)
        k32.CloseHandle(proc_handle)
        if not ok:
            k32.CloseHandle(job)
            return None

        log.info("Job Object creato (PID %s)", pid)
        return job

    except Exception as exc:
        log.warning("_assign_job_object non disponibile: %s", exc)
        return None


# ======================================================================
# TorLauncher – avvia il binario Tor su tutte le piattaforme
# ======================================================================

class TorLauncher:
    """
    Avvia il binario Tor bundled come sottoprocesso.

    Su Android:
      Il binario viene COPIATO dalla cartella dell'addon (che può essere
      su filesystem noexec) alla data-dir interna di Kodi, che è sempre
      su partizione eseguibile. Solo dopo viene avviato.

    Struttura cartelle addon:
        tor/
        ├── torrc
        └── bin/
            ├── data/
            ├── windows-x64/tor.exe
            ├── windows-x86/tor.exe
            ├── linux-x64/tor
            ├── linux-x86/tor
            ├── android-arm64/tor
            └── android-arm/tor
    """

    def __init__(self, addon_path: Optional[str] = None) -> None:
        if addon_path is None:
            addon_path = os.path.dirname(os.path.abspath(__file__))

        self.addon_path: str = addon_path
        self.tor_dir:    str = os.path.join(addon_path, "tor")
        self.torrc_path: str = os.path.join(self.tor_dir, "torrc")
        self.data_dir:   str = os.path.join(self.tor_dir, "bin", "data")
        self._process:   Optional[subprocess.Popen] = None  # type: ignore[type-arg]
        self._win_job:   Optional[object] = None

        # Popolati da _prepare_android_binary() su Android
        self._android_binary:   Optional[str] = None
        self._android_data_dir: Optional[str] = None
        self._android_torrc:    Optional[str] = None

    # ------------------------------------------------------------------
    def _get_source_binary(self) -> str:
        """Percorso del binario nell'addon (può essere su noexec)."""
        system  = platform.system().lower()
        machine = platform.machine().lower()

        if _is_android():
            exe_name = "tor"
            if machine in ("aarch64", "arm64"):
                subfolder = "android-arm64"
            elif machine.startswith("arm"):
                subfolder = "android-arm"
            elif machine in ("x86_64", "amd64"):
                subfolder = "android-x86_64"
                log.info("Android x86_64 rilevato – uso %s", subfolder)
            elif machine in ("i686", "i386", "x86"):
                subfolder = "android-x86"
                log.info("Android x86 rilevato – uso %s", subfolder)
            else:
                subfolder = "android-arm64"
                log.warning("ABI Android '%s' non riconosciuta – provo arm64", machine)
            log.info("Android – binario %s (machine=%s)", subfolder, machine)

        elif system == "windows":
            exe_name  = "tor.exe"
            subfolder = "windows-x64" if machine in ("x86_64", "amd64") else "windows-x86"

        elif system == "linux":
            exe_name  = "tor"
            subfolder = "linux-x64" if machine in ("x86_64", "amd64") else "linux-x86"

        else:
            raise RuntimeError(f"Sistema operativo non supportato: {system}")

        binary = os.path.join(self.tor_dir, "bin", subfolder, exe_name)
        if not os.path.isfile(binary):
            raise RuntimeError(
                f"Binario Tor non trovato: {binary}\n"
                "Copia il binario Tor nella cartella corretta dell'addon."
            )
        return binary

    # ------------------------------------------------------------------
    def _prepare_android_binary(self) -> bool:
        """
        Trova un percorso eseguibile, copia il binario tor e prepara torrc/data-dir.
        Prova automaticamente tutti i percorsi candidati finché ne trova uno non-noexec.
        """
        dest_dir = _find_executable_dir()
        if not dest_dir:
            log.error("_prepare_android_binary: nessun percorso eseguibile trovato")
            return False

        dest_binary   = os.path.join(dest_dir, "tor")
        dest_torrc    = os.path.join(dest_dir, "torrc")
        dest_data_dir = os.path.join(dest_dir, "data")

        try:
            os.makedirs(dest_data_dir, exist_ok=True)
        except OSError as e:
            log.error("Impossibile creare data_dir: %s", e)
            return False

        # Copia il binario (solo se mancante o aggiornato)
        try:
            source_binary = self._get_source_binary()
        except RuntimeError as e:
            log.error("_prepare_android_binary: %s", e)
            return False

        src_mtime  = os.path.getmtime(source_binary)
        dest_mtime = os.path.getmtime(dest_binary) if os.path.isfile(dest_binary) else 0

        if src_mtime > dest_mtime:
            log.info("Copia binario: %s → %s", source_binary, dest_binary)
            try:
                shutil.copy2(source_binary, dest_binary)
            except OSError as e:
                log.error("Copia binario fallita: %s", e)
                return False

        # chmod +x
        try:
            current = os.stat(dest_binary).st_mode
            if not (current & stat.S_IXUSR):
                os.chmod(dest_binary, current | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
                log.debug("chmod +x: %s", dest_binary)
        except OSError as e:
            log.error("chmod fallito: %s", e)
            return False

        # Scrive torrc con DataDirectory assoluta
        self._write_android_torrc(dest_torrc, dest_data_dir)

        self._android_binary   = dest_binary
        self._android_data_dir = dest_data_dir
        self._android_torrc    = dest_torrc

        log.info("Binario Android pronto in: %s", dest_binary)
        return True

    def _write_android_torrc(self, torrc_path: str, data_dir: str) -> None:
        """Scrive un torrc con DataDirectory assoluta per Android."""
        content = (
            f"SocksPort {TOR_SOCKS_PORT}\n"
            f"ControlPort {TOR_CTRL_PORT}\n"
            f"DataDirectory {data_dir}\n"
            "CookieAuthentication 0\n"
            "Log notice stdout\n"
        )
        # Aggiunge eventuali righe extra dal torrc dell'addon
        if os.path.isfile(self.torrc_path):
            try:
                skip_keys = ("datadirectory", "socksport", "controlport")
                with open(self.torrc_path, "r", encoding="utf-8") as f:
                    for line in f:
                        stripped = line.strip()
                        if stripped and not stripped.startswith("#"):
                            if not any(stripped.lower().startswith(k) for k in skip_keys):
                                content += line
            except Exception as e:
                log.warning("Lettura torrc addon fallita: %s", e)

        try:
            with open(torrc_path, "w", encoding="utf-8") as f:
                f.write(content)
        except OSError as e:
            log.error("Scrittura torrc fallita: %s", e)

    # ------------------------------------------------------------------
    @staticmethod
    def _ensure_executable(path: str) -> None:
        if not _is_windows():
            current = os.stat(path).st_mode
            if not (current & stat.S_IXUSR):
                try:
                    os.chmod(path, current | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
                    log.debug("chmod +x: %s", path)
                except OSError as e:
                    log.warning("chmod fallito: %s", e)

    # ------------------------------------------------------------------
    def start(self, wait_seconds: int = 25) -> bool:
        if self.is_running():
            log.info("Tor già in esecuzione (PID %s)", self._process.pid)
            return True

        if _port_open(TOR_HOST, TOR_SOCKS_PORT, timeout=1):
            log.info("Porta SOCKS %s già aperta.", TOR_SOCKS_PORT)
            return True

        # --- Android: copia il binario nella data-dir interna ---
        if _is_android():
            log.info("Android: preparo binario nella storage interna…")
            if not self._prepare_android_binary():
                log.error("Preparazione binario Android fallita.")
                return False
            binary   = self._android_binary
            torrc    = self._android_torrc
            data_dir = self._android_data_dir

        # --- Desktop: usa il binario direttamente ---
        else:
            try:
                binary = self._get_source_binary()
            except RuntimeError as exc:
                log.error("%s", exc)
                return False
            self._ensure_executable(binary)
            os.makedirs(self.data_dir, exist_ok=True)
            torrc    = self.torrc_path
            data_dir = self.data_dir

        # --- Avvio processo ---
        cmd = [binary, "-f", torrc, "--DataDirectory", data_dir]
        log.info("Avvio Tor: %s", " ".join(cmd))

        try:
            if _is_windows():
                cflags = subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
                self._process = subprocess.Popen(
                    cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                    creationflags=cflags,
                )
                self._win_job = _assign_job_object(self._process.pid)
            else:
                # Su Linux/Android: usiamo start_new_session=True al posto di
                # preexec_fn=os.setsid perché preexec_fn NON è supportato nei
                # subinterpreter di Kodi (Python 3.12+) e lancia:
                #   RuntimeError: preexec_fn not supported within subinterpreters
                #
                # start_new_session=True ha lo stesso effetto (chiama setsid()
                # nel processo figlio) ed è supportato in tutti i contesti.
                #
                # PR_SET_PDEATHSIG (SIGTERM al figlio quando il padre muore)
                # viene impostato DOPO il Popen, nel processo padre, tramite
                # un thread dedicato che usa ctypes — senza toccare preexec_fn.
                self._process = subprocess.Popen(
                    cmd,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    start_new_session=True,
                )

                # Imposta PR_SET_PDEATHSIG nel figlio subito dopo la fork,
                # usando /proc/<pid>/mem non è possibile; lo impostiamo nel
                # padre come best-effort tramite prctl sul proprio PID —
                # questo garantisce che SIGTERM venga propagato ai figli
                # quando Kodi termina (funziona su kernel Linux ≥ 3.4).
                try:
                    import ctypes
                    import signal as _signal
                    PR_SET_CHILD_SUBREAPER = 36
                    libc = ctypes.CDLL("libc.so.6", use_errno=True)
                    libc.prctl(PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0)
                    log.debug("PR_SET_CHILD_SUBREAPER impostato sul padre")
                except Exception:
                    pass   # non critico, ignora silenziosamente
        except PermissionError as exc:
            log.error("PermissionError: filesystem noexec? %s", exc)
            return False
        except Exception as exc:
            log.error("Avvio Tor fallito: %s", exc)
            return False

        log.info("Processo Tor avviato (PID %s) – attendo porta…", self._process.pid)

        deadline = time.time() + wait_seconds
        while time.time() < deadline:
            if _port_open(TOR_HOST, TOR_SOCKS_PORT, timeout=1):
                log.info("Tor pronto sulla porta %s", TOR_SOCKS_PORT)
                return True
            if self._process.poll() is not None:
                try:
                    out = self._process.stdout.read(2000).decode("utf-8", errors="replace")
                    log.error("Tor terminato prematuramente. Output:\n%s", out)
                except Exception:
                    pass
                log.error(
                    "Processo Tor terminato (exit=%s). "
                    "Su Android verifica che il binario sia compilato per la ABI corretta "
                    "(arm64 o arm).",
                    self._process.poll()
                )
                return False
            time.sleep(1)

        log.error("Tor non ha aperto la porta %s entro %s s", TOR_SOCKS_PORT, wait_seconds)
        return False

    # ------------------------------------------------------------------
    def stop(self) -> None:
        if self._process is not None:
            pid = self._process.pid
            log.info("Arresto Tor (PID %s)…", pid)
            try:
                if _is_windows():
                    subprocess.call(
                        ["taskkill", "/F", "/T", "/PID", str(pid)],
                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                    )
                    self._process.wait(timeout=5)
                else:
                    self._process.terminate()
                    self._process.wait(timeout=5)
            except Exception:
                try:
                    self._process.kill()
                except Exception:
                    pass
            finally:
                self._process = None
                self._win_job = None
                log.info("Tor arrestato.")

    def is_running(self) -> bool:
        return self._process is not None and self._process.poll() is None

    def __enter__(self) -> "TorLauncher":
        self.start()
        return self

    def __exit__(self, *_: Any) -> None:
        self.stop()


# ======================================================================
# Istanza globale launcher
# ======================================================================

_launcher: Optional[TorLauncher] = None


def get_launcher(addon_path: Optional[str] = None) -> TorLauncher:
    global _launcher
    if _launcher is None:
        _launcher = TorLauncher(addon_path=addon_path)
    return _launcher


# ======================================================================
# Singleton TorManager — UN'UNICA istanza per tutto il plugin
#
# Problema originale: functions.py, variables.py e tor_mac_check.py
# creavano ciascuno il proprio TorManager(). Questo causava:
#   - Tor avviato 3 volte (o 3 tentativi di bind su porta 9050)
#   - Sessioni requests indipendenti senza condivisione cookie/stato
#   - Overhead di inizializzazione triplicato all'avvio di Kodi
#
# Soluzione: get_tor_manager() restituisce sempre la stessa istanza.
# tutti i moduli devono importare questa funzione invece di fare
# TorManager(...) direttamente.
# ======================================================================

_tor_manager_instance: Optional["TorManager"] = None
_tor_manager_lock = threading.Lock()


def get_tor_manager(
    use_tor: bool = True,
    fallback: bool = True,
    timeout: Union[int, float] = 20,
    auto_start: bool = True,
    addon_path: Optional[str] = None,
) -> "TorManager":
    """
    Restituisce l'istanza singleton di TorManager.

    Il primo chiamante che passa addon_path lo imposta; i successivi
    ricevono l'istanza già creata indipendentemente dai parametri.
    Questo garantisce che Tor venga avviato UNA SOLA VOLTA.
    """
    global _tor_manager_instance
    if _tor_manager_instance is None:
        with _tor_manager_lock:
            # Double-checked locking: ricontrolla dentro il lock
            if _tor_manager_instance is None:
                log.info("get_tor_manager: creazione istanza singleton")
                _tor_manager_instance = TorManager(
                    use_tor=use_tor,
                    fallback=fallback,
                    timeout=timeout,
                    auto_start=auto_start,
                    addon_path=addon_path,
                )
    return _tor_manager_instance


# ======================================================================
# OrbotManager – fallback Android se il binario bundled non funziona
# ======================================================================

class OrbotManager:
    """
    Fallback di secondo livello: se il binario bundled fallisce su Android,
    prova a usare il proxy SOCKS5 di Orbot (se installato e avviato).
    """
    ORBOT_PORTS = [9050, 9150]

    def __init__(self) -> None:
        self._active_port: Optional[int] = None

    def detect_port(self) -> Optional[int]:
        for port in self.ORBOT_PORTS:
            if _port_open(TOR_HOST, port, timeout=2):
                log.info("Orbot rilevato sulla porta %s", port)
                self._active_port = port
                return port
        return None

    def is_available(self) -> bool:
        return self.detect_port() is not None

    def get_proxy_url(self) -> Optional[str]:
        port = self._active_port or self.detect_port()
        return f"socks5h://{TOR_HOST}:{port}" if port else None

    @staticmethod
    def show_dialog() -> None:
        try:
            import xbmcgui
            xbmcgui.Dialog().ok(
                "Tor non disponibile",
                (
                    "Impossibile avviare Tor su questo dispositivo Android.\n\n"
                    "Come alternativa puoi installare [B]Orbot[/B] "
                    "(l'app Tor ufficiale per Android), avviarla e "
                    "riaprire l'addon.\n\n"
                    "L'addon continuerà in modalità connessione diretta."
                )
            )
        except Exception:
            pass


# ======================================================================
# Rilevamento pagine di blocco (AGCOM, geo-restriction, ecc.)
# ======================================================================

_BLOCK_MARKERS = [
    "accesso al presente sito",
    "disabilitato in ottemperanza",
    "autorita per le garanzie",
    "agcom",
    "this site has been blocked",
    "access to this website has been blocked",
    "blocked by your internet service provider",
    "il tuo provider ha bloccato",
    "not available in your country",
    "not available in your region",
    "geo-restricted",
    "content is not available",
]


def _is_blocked_response(text: str) -> bool:
    """
    True se il testo sembra una pagina di blocco AGCOM/ISP/geo-restriction.
    """
    if not text:
        return True
    if len(text) > 50000:
        return False
    lower = text.lower()
    return any(marker in lower for marker in _BLOCK_MARKERS)


# ======================================================================
# TorManager – wrapper requests.Session multipiattaforma
# ======================================================================

class TorManager:
    """
    Wrapper su requests.Session.

    Ordine di tentativi su Android:
      1. Avvia binario Tor bundled (copiato nella data-dir interna)
      2. Se fallisce → prova Orbot
      3. Se fallisce → connessione diretta (se fallback=True)

    Su Windows/Linux:
      1. Avvia binario Tor bundled
      2. Se fallisce → connessione diretta (se fallback=True)
    """

    def __init__(
        self,
        use_tor:    bool = True,
        fallback:   bool = True,
        timeout:    Union[int, float] = 20,
        auto_start: bool = True,
        addon_path: Optional[str] = None,
    ) -> None:
        self.use_tor         = use_tor
        self.fallback        = fallback
        self.default_timeout = timeout
        self._tor_ready      = False
        self._proxy_url: Optional[str] = None

        if use_tor:
            self._setup_tor(auto_start=auto_start, addon_path=addon_path)

        self.session = requests.Session()
        self.session.headers.update(DEFAULT_HEADERS)

        if self._proxy_url:
            self.session.proxies.update({
                "http":  self._proxy_url,
                "https": self._proxy_url,
            })
            log.debug("Sessione → proxy %s", self._proxy_url)
        elif use_tor:
            log.warning("Tor/Orbot non disponibile – connessione diretta.")

    def _setup_tor(self, auto_start: bool, addon_path: Optional[str]) -> None:
        # ── Porta già aperta (Tor/APK già attivo da sessione precedente) ──────
        # Controlla prima di tutto: se la porta è aperta non serve riavviare nulla.
        if _port_open(TOR_HOST, TOR_SOCKS_PORT, timeout=1):
            log.info("Porta SOCKS %s già aperta – riuso sessione esistente.", TOR_SOCKS_PORT)
            self._proxy_url = TOR_PROXY
            self._tor_ready = True
            return

        if not auto_start:
            # auto_start=False e porta chiusa → connessione diretta
            return

        # ── Android ──────────────────────────────────────────────────────────
        if _is_android():
            # Strategia 1: binario Tor bundled (solo se filesystem non è noexec)
            if not _android_confirmed_all_noexec:
                log.info("Android: provo binario Tor bundled…")
                launcher = get_launcher(addon_path=addon_path)
                ok = launcher.start()
                if ok:
                    self._proxy_url = TOR_PROXY
                    self._tor_ready = True
                    return
                log.warning("Android: binario bundled fallito.")
            else:
                log.debug(
                    "Android: tutti noexec (flag persistente) – salto binario bundled"
                )

            # Strategia 2: TorProxy APK esterno (NON-BLOCCANTE).
            #
            # PROBLEMA PRECEDENTE: _launch_torproxy_apk() attendeva 60+ secondi
            # bloccando il thread principale. Kodi killa gli script dopo ~5 s →
            # "script didn't stop in 5 seconds - let's kill it".
            #
            # SOLUZIONE:
            #   a) StartAndroidActivity (API Kodi nativa) invece di "am" da shell
            #      (am restituiva sempre codice 255 per mancanza di permessi).
            #   b) L'attesa della porta avviene in un thread daemon separato.
            #      Il thread aggiorna self._proxy_url e self.session.proxies
            #      quando Tor diventa disponibile, senza bloccare Kodi.
            log.info(
                "Android: lancio TorProxy APK (%s) in background…", TORPROXY_APK_PACKAGE
            )
            sent = _launch_torproxy_apk_nonblocking()
            if sent:
                # Controlla subito se era già attivo (avvio istantaneo)
                if _port_open(TOR_HOST, TOR_SOCKS_PORT, timeout=2):
                    self._proxy_url = TOR_PROXY
                    self._tor_ready = True
                    log.info("Android: TorProxy APK già attivo su %s", TOR_PROXY)
                    return
                # Avvio in background: monitora la porta senza bloccare Kodi
                self._start_tor_watcher()
                log.info(
                    "Android: APK avviata – attesa porta %s in background thread.",
                    TOR_SOCKS_PORT,
                )
                # Restituisce il controllo a Kodi subito; il watcher aggiornerà
                # la sessione quando Tor sarà pronto. Nel frattempo si usa
                # connessione diretta/bypass (se fallback=True).
                return
            log.warning("Android: impossibile avviare TorProxy APK.")

            # Strategia 3: Orbot (se installato e già avviato dall'utente)
            orbot = OrbotManager()
            if orbot.is_available():
                proxy_url = orbot.get_proxy_url()
                log.info("Android: Orbot rilevato su %s", proxy_url)
                self._proxy_url = proxy_url
                self._tor_ready = True
                return

            # Strategia 4: proxy configurato manualmente nelle impostazioni
            user_proxy = _get_user_proxy()
            if user_proxy:
                log.info("Uso proxy utente: %s", user_proxy)
                self._proxy_url = user_proxy
                self._tor_ready = True
                return

            # Nessuna opzione disponibile
            log.warning(
                "Android: nessun tunnel Tor disponibile. "
                "Connessione diretta (bypass DoH attivo)."
            )
            self._tor_ready = False
            return

        # Desktop (Windows/Linux)
        launcher = get_launcher(addon_path=addon_path)
        ok = launcher.start()
        if ok:
            self._proxy_url = TOR_PROXY
            self._tor_ready = True
            return

        self._tor_ready = False

    # ------------------------------------------------------------------
    def _start_tor_watcher(self, max_wait: int = 90) -> None:
        """
        Avvia un thread daemon che monitora la porta SOCKS di Tor.
        Quando la porta si apre, aggiorna self._proxy_url e self.session.proxies
        in modo thread-safe, senza bloccare il thread principale di Kodi.

        max_wait: secondi massimi di attesa (default 90).
        """
        def _watch() -> None:
            log.info("TorWatcher: avvio – attendo porta %s (max %s s)…",
                     TOR_SOCKS_PORT, max_wait)
            deadline = time.time() + max_wait
            while time.time() < deadline:
                if _port_open(TOR_HOST, TOR_SOCKS_PORT, timeout=1):
                    log.info("TorWatcher: porta %s aperta – attivo proxy Tor.",
                             TOR_SOCKS_PORT)
                    self._proxy_url = TOR_PROXY
                    self._tor_ready = True
                    # Aggiorna la sessione requests con il proxy
                    try:
                        self.session.proxies.update({
                            "http":  TOR_PROXY,
                            "https": TOR_PROXY,
                        })
                        log.info("TorWatcher: sessione aggiornata con proxy %s", TOR_PROXY)
                    except Exception as exc:
                        log.error("TorWatcher: errore aggiornamento sessione: %s", exc)
                    return
                time.sleep(2)
            log.warning(
                "TorWatcher: porta %s non aperta entro %s s. "
                "Continuazione in modalità connessione diretta.",
                TOR_SOCKS_PORT, max_wait,
            )

        t = threading.Thread(target=_watch, name="TorWatcher", daemon=True)
        t.start()

    # ------------------------------------------------------------------
    def get(
        self,
        url: str,
        timeout: Optional[Union[int, float]] = None,
        headers: Optional[Dict[str, str]] = None,
    ) -> Optional[str]:
        timeout     = timeout if timeout is not None else self.default_timeout
        req_headers = {**DEFAULT_HEADERS, **(headers or {})}

        # Android senza Tor/proxy: bypass diretto — i server IPTV
        # rifiutano la connessione diretta da Android, inutile provarla.
        if _is_android() and not self._tor_ready:
            log.debug("Android senza proxy – bypass diretto per: %s", url)
            return self._try_bypass(url, req_headers, timeout)

        # Tentativo via session (proxy Tor/SOCKS5/HTTP configurato sulla sessione).
        # NON usiamo requests.get() globale per evitare che eventuali
        # modifiche residue al socket globale interferiscano.
        if self._proxy_url:
            try:
                log.debug("GET via proxy session: %s", url)
                resp = self.session.get(
                    url, headers=req_headers, timeout=timeout, verify=False
                )
                if resp.status_code < 400:
                    log.debug("GET proxy ok – %d byte", len(resp.content))
                    return resp.text
                log.warning("GET proxy HTTP %s per: %s", resp.status_code, url)
            except Exception as err:
                log.error("GET via proxy fallito: %s", err)
            if not self.fallback:
                return None
            log.info("FALLBACK bypass per: %s", url)
            return self._try_bypass(url, req_headers, timeout)

        # Desktop senza proxy (Tor non avviato) — usa la session
        # senza proxies configurati, poi bypass se bloccata.
        try:
            log.debug("GET diretto (session): %s", url)
            resp = self.session.get(url, headers=req_headers, timeout=timeout)
            if resp.status_code < 400:
                text = resp.text
                if _is_blocked_response(text):
                    log.warning("Risposta bloccata – provo bypass")
                else:
                    return text
        except Exception as err2:
            log.warning("GET diretto fallito: %s – provo bypass", err2)

        return self._try_bypass(url, req_headers, timeout)

    def _try_bypass(
        self,
        url: str,
        req_headers: Dict,
        timeout: Union[int, float],
    ) -> Optional[str]:
        """Unico punto di accesso a bypass.py in tutto TorManager."""
        try:
            from bypass import get_with_bypass
            log.debug("_try_bypass per: %s", url)
            result = get_with_bypass(url, headers=req_headers, timeout=int(timeout))
            if result and not _is_blocked_response(result):
                log.info("Bypass OK: %s", url)
                return result
            if result:
                log.debug("Bypass: risposta ricevuta")
                return result
            log.warning("Bypass: nessuna risposta per: %s", url)
        except Exception as e:
            log.error("Bypass errore: %s", e)
        return None

    # ------------------------------------------------------------------
    def get_json(
        self,
        url: str,
        timeout: Optional[Union[int, float]] = None,
        headers: Optional[Dict[str, str]] = None,
    ) -> Optional[Any]:
        timeout     = timeout if timeout is not None else self.default_timeout
        req_headers = {**DEFAULT_HEADERS, **(headers or {})}

        if self.use_tor and self._tor_ready:
            try:
                log.debug("GET-JSON via Tor: %s", url)
                resp = self.session.get(url, headers=req_headers, timeout=timeout)
                resp.raise_for_status()
                return resp.json()
            except Exception as err:
                log.error("Tor GET-JSON fallito: %s", err)
                if not self.fallback:
                    return None
                log.info("FALLBACK: GET-JSON diretto")

        try:
            resp = requests.get(url, headers=req_headers, timeout=timeout)
            resp.raise_for_status()
            return resp.json()
        except Exception as err2:
            log.error("Fallback GET-JSON fallito: %s", err2)
            return None


# ======================================================================
# Singleton TorManager — unica istanza condivisa da tutto il plugin
# ======================================================================
# Funzioni di utilità pubbliche
# ======================================================================

def rotate_ip() -> None:
    """Invia SIGNAL NEWNYM a Tor per cambiare circuito."""
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(5)
        s.connect((TOR_HOST, TOR_CTRL_PORT))
        s.sendall(b'AUTHENTICATE ""\r\n')
        resp = s.recv(1024)
        if not resp.startswith(b'250'):
            raise RuntimeError(f"Auth fallita: {resp!r}")
        s.sendall(b'SIGNAL NEWNYM\r\n')
        resp = s.recv(1024)
        if not resp.startswith(b'250'):
            raise RuntimeError(f"NEWNYM fallito: {resp!r}")
        s.close()
        log.info("IP Tor ruotato – attesa 5 s")
        time.sleep(5)
    except Exception as exc:
        log.error("Rotate IP fallito: %s", exc)


def test_tor() -> str:
    """Verifica la connessione Tor su check.torproject.org."""
    active_port = None
    for p in [TOR_SOCKS_PORT, 9150]:
        if _port_open(TOR_HOST, p, timeout=2):
            active_port = p
            break

    if active_port is None:
        return "TOR TEST FAILED: nessun proxy SOCKS5 su 9050/9150"

    proxy_url = f"socks5h://{TOR_HOST}:{active_port}"
    try:
        resp = requests.get(
            "https://check.torproject.org/api/ip",
            proxies={"http": proxy_url, "https": proxy_url},
            timeout=10,
        )
        return resp.text
    except Exception as exc:
        return f"TOR TEST FAILED: {exc}"


# ======================================================================
# Mini-test da riga di comando
# ======================================================================
if __name__ == "__main__":
    print(f"[tor_manager] Piattaforma: {platform.system()} | Android: {_is_android()}")
    launcher = TorLauncher()
    if launcher.start():
        print("[tor_manager] Tor avviato – test connessione…")
        print(test_tor())
        tm = TorManager(auto_start=False)
        html = tm.get("https://httpbin.org/html")
        if html:
            print("\nPrime 200 caratteri:")
            print(html[:200])
        launcher.stop()
    else:
        print("[tor_manager] Tor NON avviato.")
        if _is_android():
            print("Verifica ABI: android-arm64/tor o android-arm/tor")
