sp-hydra-veil-cli/cli/ui.py

import sys
import threading
import time

RESET   = "\033[0m"
BOLD    = "\033[1m"
DIM     = "\033[2m"

RED     = "\033[0;31m"
GREEN   = "\033[0;32m"
YELLOW  = "\033[0;33m"
BLUE    = "\033[0;34m"
CYAN    = "\033[0;36m"
WHITE   = "\033[0;37m"

BRED    = "\033[1;31m"
BGREEN  = "\033[1;32m"
BYELLOW = "\033[1;33m"
BBLUE   = "\033[1;34m"
BCYAN   = "\033[1;36m"
BWHITE  = "\033[1;37m"

def label_ok(msg: str):
    print(f"  {BGREEN}[  OK  ]{RESET}  {msg}")

def label_warn(msg: str):
    print(f"  {BYELLOW}[ WARN ]{RESET}  {msg}")

def label_error(msg: str):
    print(f"  {BRED}[ ERROR]{RESET}  {msg}")

def label_info(msg: str):
    print(f"  {BCYAN}[ INFO ]{RESET}  {msg}")

def label_skip(msg: str):
    print(f"  {DIM}[ SKIP ]{RESET}  {msg}")

def label_connected(msg: str):
    print(f"  {BGREEN}[CONNECTED   ]{RESET}  {msg}")

def label_disconnect(msg: str):
    print(f"  {BRED}[DISCONNECTED]{RESET}  {msg}")

def label_step(msg: str, current: int, total: int):
    print(f"  {BBLUE}[ {current:<2}/{total:<2} ]{RESET}  {msg}")

def print_header(title: str, version: str = ""):
    width = 60
    line = "─" * width
    ver = f"{DIM}{version}{RESET}" if version else ""
    print()
    print(f"{BCYAN}{line}{RESET}")
    print(f"{BCYAN}  {'':<56}{RESET}")
    print(f"{BCYAN}  {BWHITE}{title} {ver}{'':<40}{BCYAN}  {RESET}")
    print(f"{BCYAN}  {'':<56}{RESET}")
    print(f"{BCYAN}{line}{RESET}")
    print()

def print_section(label: str):
    print()
    print(f"{DIM}{CYAN}┌─{RESET} {BWHITE}{label}{RESET}")

def print_divider():
    print(f"{DIM}{'─' * 60}{RESET}")

def progress_bar(current: int, total: int, label: str = "", width: int = 40):
    filled  = int(current * width / total)
    empty   = width - filled
    bar     = "#" * filled + "·" * empty
    pct     = int(current * 100 / total)
    sys.stdout.write(
        f"\r  {DIM}[{RESET}{BGREEN}{bar}{RESET}{DIM}]{RESET}"
        f" {BWHITE}{pct:3d}%{RESET}  {DIM}{label}{RESET}   "
    )
    sys.stdout.flush()

def progress_simulate(steps: int = 20, label: str = "loading", delay: float = 0.04):
    for i in range(1, steps + 1):
        progress_bar(i, steps, label)
        time.sleep(delay)
    print()

class Spinner:
    FRAMES = ("─", "\\", "│", "/")

    def __init__(self, msg: str = "Processing..."):
        self.msg     = msg
        self._stop_event = threading.Event()
        self._thread: threading.Thread | None = None

    def start(self):
        self._stop_event.clear()
        self._thread = threading.Thread(target=self._spin, daemon=True)
        self._thread.start()
        return self

    def stop(self, result_msg: str = ""):
        self._stop_event.set()
        if self._thread:
            self._thread.join()
        sys.stdout.write(f"\r{' ' * 60}\r")
        sys.stdout.flush()
        if result_msg:
            label_ok(result_msg)

    def _spin(self):
        i = 0
        while not self._stop_event.is_set():
            frame = self.FRAMES[i % len(self.FRAMES)]
            sys.stdout.write(
                f"\r  {BCYAN}{frame}{RESET}  {DIM}{self.msg}{RESET}   "
            )
            sys.stdout.flush()
            i += 1
            time.sleep(0.1)

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

    def __exit__(self, *_):
        self.stop()


def wait_with_spinner(msg: str, fn, *args, result_msg: str = "", **kwargs):
    result_box = [None, None]

    def _run():
        try:
            result_box[0] = fn(*args, **kwargs)
        except Exception as e:
            result_box[1] = e

    s = Spinner(msg)
    s.start()
    t = threading.Thread(target=_run, daemon=True)
    t.start()
    t.join()
    s.stop(result_msg)

    if result_box[1] is not None:
        raise result_box[1]
    return result_box[0]


def label_dns(enabled: bool, active: bool):
    if not enabled:
        print(f"  {BYELLOW}[ DNS ]{RESET}  DNS management {BYELLOW}disabled{RESET} — using system resolver")
    elif active:
        print(f"  {BGREEN}[ DNS ]{RESET}  DNS over {BGREEN}tun0{RESET} active  {DIM}(9.9.9.9){RESET}")
    else:
        print(f"  {BRED}[ DNS ]{RESET}  DNS on tun0 {BRED}FAILED{RESET} — possible DNS leak")


def label_killswitch(armed: bool):
    if armed:
        print(f"  {BGREEN}[  KS  ]{RESET}  Kill switch active")
    else:
        print(f"  {BRED}[  KS  ]{RESET}  Kill switch {BRED}INACTIVE — connection unprotected{RESET}")


def session_timer(drop_state: dict | None = None, tunnel_state: dict | None = None) -> threading.Event:

    stop_event = threading.Event()
    start      = time.time()

    def _run():
        while not stop_event.is_set():
            elapsed = int(time.time() - start)
            h = elapsed // 3600
            m = (elapsed % 3600) // 60
            s = elapsed % 60
            time_str = f"{h:02d}:{m:02d}:{s:02d}"

            alive = True if tunnel_state is None else tunnel_state.get("alive", True)
            dot_color = BGREEN if alive else BRED
            status_word = "Protected" if alive else "Disconnected"

            if drop_state is not None:
                count = drop_state["count"]
                drop_color = BRED if count > 0 else DIM
                drop_str = (
                    f"  {DIM}│{RESET}  "
                    f"{drop_color}drops: {count}{RESET}"
                )
            else:
                drop_str = ""

            sys.stdout.write(
                f"\r  {dot_color}●{RESET}  {BWHITE}{status_word}{RESET}  "
                f"{DIM}[{time_str}]{RESET}"
                f"{drop_str}"
                f"   "
            )
            sys.stdout.flush()
            time.sleep(1)

        sys.stdout.write("\r" + " " * 70 + "\r")
        sys.stdout.flush()

    threading.Thread(target=_run, daemon=True).start()
    return stop_event


def connecting_spinner() -> threading.Event:
    stop_event = threading.Event()
    frames = ("─", "\\", "│", "/")

    def _run():
        i = 0
        while not stop_event.is_set():
            frame = frames[i % len(frames)]
            sys.stdout.write(
                f"\r  {BYELLOW}{frame}{RESET}  {DIM}Reconnecting...{RESET}   "
            )
            sys.stdout.flush()
            i += 1
            time.sleep(0.12)
        sys.stdout.write("\r" + " " * 60 + "\r")
        sys.stdout.flush()

    threading.Thread(target=_run, daemon=True).start()
    return stop_event