"""Home-tunnel proxy control panel for the TutorialMaker Space. Runs a local authenticated HTTP forward proxy on your machine (your residential IP) and exposes it to the internet through a RAW TCP tunnel (ngrok / bore), so the Hugging Face Space can route youtube-transcript-api + yt-dlp through your home IP and dodge the datacenter-IP block. One-click sets the resulting URL as the Space's YT_PROXY secret. Why a TCP tunnel: a forward proxy speaks HTTP CONNECT, which only survives a transparent TCP tunnel. HTTP reverse tunnels (Cloudflare quick-tunnel, Tailscale Funnel) do NOT work for this — don't use them here. Run: python tools/home_proxy_panel.py Deps: pip install proxy.py huggingface_hub requests (ngrok or bore on PATH for tunneling) """ from __future__ import annotations import json import os import platform import queue import re import secrets import shutil import subprocess import sys import tarfile import threading import zipfile CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".tutorialmaker_proxy.json") DEFAULT_REPO = "vivekchakraverty/TutorialMaker" _NO_WINDOW = 0x08000000 if os.name == "nt" else 0 # Where to keep a downloaded bore binary. When frozen by PyInstaller sys.frozen is set and # the binary lives next to the executable; otherwise next to this script. APP_DIR = os.path.dirname(sys.executable if getattr(sys, "frozen", False) else os.path.abspath(__file__)) def _bore_target_triple() -> tuple[str, str, str] | None: """(rust-triple, archive-ext, binary-name) for the current OS/arch, or None.""" sysname = platform.system() mach = platform.machine().lower() is_arm = mach in ("arm64", "aarch64") if sysname == "Windows": return ("x86_64-pc-windows-msvc", ".zip", "bore.exe") if sysname == "Darwin": triple = "aarch64-apple-darwin" if is_arm else "x86_64-apple-darwin" return (triple, ".tar.gz", "bore") if sysname == "Linux": triple = "aarch64-unknown-linux-musl" if is_arm else "x86_64-unknown-linux-musl" return (triple, ".tar.gz", "bore") return None def ensure_bore(log=lambda m: None) -> str | None: """Return a runnable bore path: PATH copy, a previously downloaded one, or a freshly downloaded release binary for this OS/arch. Returns None if it can't be obtained.""" found = shutil.which("bore") if found: return found tri = _bore_target_triple() if not tri: log("[bore] no prebuilt binary for this platform — install bore manually.") return None triple, ext, binname = tri local = os.path.join(APP_DIR, binname) if os.path.exists(local): return local import urllib.request api = "https://api.github.com/repos/ekzhang/bore/releases/latest" try: log("[bore] not found — fetching latest release info…") req = urllib.request.Request(api, headers={"User-Agent": "tutorialmaker-panel"}) with urllib.request.urlopen(req, timeout=30) as r: rel = json.load(r) asset = next((a for a in rel.get("assets", []) if triple in a["name"] and a["name"].endswith(ext)), None) if not asset: log(f"[bore] no asset matching {triple}{ext} in the latest release.") return None url = asset["browser_download_url"] archive = os.path.join(APP_DIR, asset["name"]) log(f"[bore] downloading {asset['name']}…") dl = urllib.request.Request(url, headers={"User-Agent": "tutorialmaker-panel"}) with urllib.request.urlopen(dl, timeout=120) as resp, open(archive, "wb") as fh: shutil.copyfileobj(resp, fh) if ext == ".zip": with zipfile.ZipFile(archive) as z: z.extract(binname, APP_DIR) else: with tarfile.open(archive) as t: member = next((m for m in t.getmembers() if os.path.basename(m.name) == binname), None) if member is None: log("[bore] archive did not contain the bore binary.") return None member.name = binname t.extract(member, APP_DIR) try: os.remove(archive) except OSError: pass if os.name != "nt": os.chmod(local, 0o755) log(f"[bore] ready at {local}") return local except Exception as exc: log(f"[bore] download failed: {type(exc).__name__}: {str(exc)[:160]}") return None # ----------------------------------------------------------------- pure helpers (tested) def build_proxy_url(host: str, port, user: str, pw: str) -> str: """Compose an http proxy URL, url-encoding the credentials.""" from urllib.parse import quote auth = "" if user or pw: auth = f"{quote(user, safe='')}:{quote(pw, safe='')}@" return f"http://{auth}{host}:{port}" def parse_ngrok_tunnels(json_text: str) -> str | None: """Return 'host:port' from ngrok's /api/tunnels JSON (first tcp tunnel).""" try: data = json.loads(json_text) except json.JSONDecodeError: return None for t in data.get("tunnels", []): pub = t.get("public_url", "") if pub.startswith("tcp://"): return pub[len("tcp://"):] return None def parse_bore_endpoint(text: str) -> str | None: """Return 'bore.pub:PORT' parsed from bore's log output.""" m = re.search(r"(bore\.pub:\d+)", text) if m: return m.group(1) m = re.search(r"listening at\s+([\w.-]+:\d+)", text) return m.group(1) if m else None def hf_candidate_tokens() -> list[str]: """HF tokens to try: cached token, then any huggingface.co git-credential.""" out, seen = [], set() env = dict(os.environ) env.pop("HF_TOKEN", None) env.pop("HUGGING_FACE_HUB_TOKEN", None) try: from huggingface_hub import get_token # get_token reads HF_TOKEN env first; temporarily clear it for the cached one saved = os.environ.pop("HF_TOKEN", None) try: t = get_token() finally: if saved is not None: os.environ["HF_TOKEN"] = saved if t and t not in seen: seen.add(t); out.append(t) except Exception: pass cred = os.path.join(os.path.expanduser("~"), ".git-credentials") if os.path.exists(cred): with open(cred, encoding="utf-8", errors="ignore") as fh: for line in fh: m = re.search(r"https://[^:]*:([^@]+)@huggingface\.co", line) if m and m.group(1) not in seen: seen.add(m.group(1)); out.append(m.group(1)) return out def set_space_secret(repo: str, key: str, value: str, token: str | None = None) -> str: """Set a Space secret; return the authenticated username. Raises on failure.""" from huggingface_hub import HfApi tokens = [token] if token else hf_candidate_tokens() last = None for tok in tokens: try: api = HfApi(token=tok) who = api.whoami()["name"] api.add_space_secret(repo_id=repo, key=key, value=value) return who except Exception as exc: last = exc raise RuntimeError(f"Could not set secret: {last}") def test_through_proxy(proxy_url: str) -> dict: """Route a couple of requests through proxy_url; report egress IP + YouTube reach.""" import requests proxies = {"http": proxy_url, "https": proxy_url} out = {} try: out["egress_ip"] = requests.get("https://api.ipify.org", proxies=proxies, timeout=25).text.strip() except Exception as exc: out["egress_ip"] = f"FAILED: {type(exc).__name__}: {str(exc)[:120]}" try: r = requests.get("https://www.youtube.com/robots.txt", proxies=proxies, timeout=25) out["youtube"] = f"reachable (HTTP {r.status_code})" except Exception as exc: out["youtube"] = f"FAILED: {type(exc).__name__}: {str(exc)[:120]}" return out # ------------------------------------------------------------------- subprocess manager class Proc: """A subprocess whose stdout/stderr lines are streamed into a queue.""" def __init__(self, log_q: "queue.Queue", tag: str): self.p = None self.log_q = log_q self.tag = tag def start(self, cmd: list[str]): self.stop() self.log_q.put(f"[{self.tag}] $ " + " ".join(cmd)) self.p = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, creationflags=_NO_WINDOW) threading.Thread(target=self._pump, daemon=True).start() def _pump(self): for line in self.p.stdout: self.log_q.put(f"[{self.tag}] " + line.rstrip()) def running(self) -> bool: return self.p is not None and self.p.poll() is None def stop(self): if self.running(): try: self.p.terminate() except Exception: pass self.p = None # ---------------------------------------------------------------------------- GUI shell def main(): import tkinter as tk from tkinter import ttk, messagebox cfg = {} if os.path.exists(CONFIG_PATH): try: cfg = json.load(open(CONFIG_PATH, encoding="utf-8")) except Exception: cfg = {} root = tk.Tk() root.title("TutorialMaker — Home Proxy Panel") root.geometry("760x620") log_q: "queue.Queue" = queue.Queue() proxy_proc = Proc(log_q, "proxy") tunnel_proc = Proc(log_q, "tunnel") state = {"endpoint": None} def v(name, default=""): return tk.StringVar(value=str(cfg.get(name, default))) port = v("port", "8899") user = v("user", "tm") pw = v("pw", secrets.token_urlsafe(10)) provider = v("provider", "ngrok") ngrok_token = v("ngrok_token", "") manual_ep = v("manual_ep", "") repo = v("repo", DEFAULT_REPO) hf_token = v("hf_token", "") endpoint_var = tk.StringVar(value="(not running)") yturl_var = tk.StringVar(value="") status_var = tk.StringVar(value="idle") def log(msg): log_q.put(msg) def save_cfg(): json.dump({"port": port.get(), "user": user.get(), "pw": pw.get(), "provider": provider.get(), "ngrok_token": ngrok_token.get(), "manual_ep": manual_ep.get(), "repo": repo.get()}, open(CONFIG_PATH, "w", encoding="utf-8")) log(f"[cfg] saved to {CONFIG_PATH}") def compute_yturl(): ep = state["endpoint"] if not ep: yturl_var.set("") return None host, _, p = ep.partition(":") url = build_proxy_url(host, p, user.get(), pw.get()) yturl_var.set(url) return url # --- proxy --- def start_proxy(): auth = f"{user.get()}:{pw.get()}" pargs = ["--hostname", "127.0.0.1", "--port", port.get(), "--basic-auth", auth] if getattr(sys, "frozen", False): # In a PyInstaller build sys.executable is THIS app, not python, so `-m proxy` # won't work — re-invoke ourselves with the --run-proxy sentinel (see __main__). cmd = [sys.executable, "--run-proxy", *pargs] else: cmd = [sys.executable, "-m", "proxy", *pargs] proxy_proc.start(cmd) status_var.set("proxy starting…") def stop_proxy(): proxy_proc.stop(); log("[proxy] stopped") # --- tunnel --- def start_tunnel(): prov = provider.get() if prov == "manual": ep = manual_ep.get().strip() if not ep: messagebox.showwarning("Manual", "Enter host:port for the manual endpoint.") return state["endpoint"] = ep endpoint_var.set(ep); compute_yturl(); log(f"[tunnel] manual endpoint {ep}") return if prov == "ngrok": if ngrok_token.get().strip(): subprocess.run(["ngrok", "config", "add-authtoken", ngrok_token.get().strip()], creationflags=_NO_WINDOW) tunnel_proc.start(["ngrok", "tcp", port.get(), "--log", "stdout"]) root.after(2500, _grab_ngrok) elif prov == "bore": status_var.set("preparing bore…") def work(): bore = ensure_bore(log) if not bore: log("[tunnel] bore unavailable — download bore.exe from " "https://github.com/ekzhang/bore/releases or pick ngrok.") status_var.set("bore unavailable") return root.after(0, lambda: tunnel_proc.start( [bore, "local", port.get(), "--to", "bore.pub"])) status_var.set("tunnel starting…") threading.Thread(target=work, daemon=True).start() return status_var.set("tunnel starting…") def _grab_ngrok(): try: import requests j = requests.get("http://127.0.0.1:4040/api/tunnels", timeout=5).text ep = parse_ngrok_tunnels(j) if ep: state["endpoint"] = ep; endpoint_var.set(ep); compute_yturl() log(f"[tunnel] ngrok endpoint {ep}") else: log("[tunnel] ngrok: no tcp tunnel yet, retrying…"); root.after(2000, _grab_ngrok) except Exception as exc: log(f"[tunnel] ngrok api: {exc}"); root.after(2000, _grab_ngrok) def stop_tunnel(): tunnel_proc.stop(); state["endpoint"] = None endpoint_var.set("(not running)"); yturl_var.set(""); log("[tunnel] stopped") # --- actions --- def copy_url(): url = compute_yturl() if url: root.clipboard_clear(); root.clipboard_append(url); log("[yturl] copied") def do_test(): url = compute_yturl() if not url: messagebox.showinfo("Test", "Start the tunnel first."); return status_var.set("testing…") def work(): res = test_through_proxy(url) log(f"[test] egress IP: {res['egress_ip']}") log(f"[test] youtube: {res['youtube']}") status_var.set("test done") threading.Thread(target=work, daemon=True).start() def do_set_secret(): url = compute_yturl() if not url: messagebox.showinfo("Secret", "Start the tunnel first."); return status_var.set("setting YT_PROXY…") def work(): try: who = set_space_secret(repo.get().strip(), "YT_PROXY", url, hf_token.get().strip() or None) log(f"[hf] YT_PROXY set on {repo.get()} (as {who})") status_var.set("YT_PROXY set ✓") except Exception as exc: log(f"[hf] FAILED: {exc}"); status_var.set("secret failed") threading.Thread(target=work, daemon=True).start() # --- layout --- pad = {"padx": 6, "pady": 3} frm = ttk.Frame(root); frm.pack(fill="x", **pad) r = 0 def row(label, var, show=None, width=42): nonlocal r ttk.Label(frm, text=label).grid(row=r, column=0, sticky="w", **pad) e = ttk.Entry(frm, textvariable=var, width=width, show=show) e.grid(row=r, column=1, columnspan=3, sticky="w", **pad); r += 1 return e row("Proxy port", port, width=10) row("Proxy user", user, width=18) row("Proxy password", pw, width=24) ttk.Label(frm, text="Tunnel provider").grid(row=r, column=0, sticky="w", **pad) ttk.Combobox(frm, textvariable=provider, values=["ngrok", "bore", "manual"], width=10, state="readonly").grid(row=r, column=1, sticky="w", **pad); r += 1 row("ngrok authtoken (ngrok only)", ngrok_token, show="*") row("Manual endpoint host:port (manual only)", manual_ep, width=28) row("HF Space repo", repo, width=34) row("HF token (blank = auto-detect)", hf_token, show="*") btns = ttk.Frame(root); btns.pack(fill="x", **pad) for txt, fn in [("Start proxy", start_proxy), ("Stop proxy", stop_proxy), ("Start tunnel", start_tunnel), ("Stop tunnel", stop_tunnel)]: ttk.Button(btns, text=txt, command=fn).pack(side="left", **pad) btns2 = ttk.Frame(root); btns2.pack(fill="x", **pad) for txt, fn in [("📋 Copy my Proxy URL", copy_url), ("Test proxy", do_test), ("Save config", save_cfg), ("New password", lambda: pw.set(secrets.token_urlsafe(10))), ("Set on Space (owner)", do_set_secret)]: ttk.Button(btns2, text=txt, command=fn).pack(side="left", **pad) info = ttk.Frame(root); info.pack(fill="x", **pad) ttk.Label(info, text="Public endpoint:").grid(row=0, column=0, sticky="w", **pad) ttk.Label(info, textvariable=endpoint_var, foreground="#0a7").grid(row=0, column=1, sticky="w", **pad) ttk.Label(info, text="Your proxy URL:").grid(row=1, column=0, sticky="w", **pad) ttk.Entry(info, textvariable=yturl_var, width=70, state="readonly").grid(row=1, column=1, sticky="w", **pad) ttk.Label(info, text="Status:").grid(row=2, column=0, sticky="w", **pad) ttk.Label(info, textvariable=status_var, foreground="#06c").grid(row=2, column=1, sticky="w", **pad) txt = tk.Text(root, height=16, wrap="word", bg="#111", fg="#ddd") txt.pack(fill="both", expand=True, **pad) def poll(): try: while True: line = log_q.get_nowait() txt.insert("end", line + "\n"); txt.see("end") except queue.Empty: pass # refresh running status s = [] s.append("proxy:on" if proxy_proc.running() else "proxy:off") s.append("tunnel:on" if tunnel_proc.running() else "tunnel:off") if status_var.get() in ("idle", "proxy starting…", "tunnel starting…"): status_var.set(" ".join(s)) root.after(400, poll) def on_close(): proxy_proc.stop(); tunnel_proc.stop(); root.destroy() root.protocol("WM_DELETE_WINDOW", on_close) log("Ready. 1) Start proxy 2) Start tunnel 3) Test proxy 4) Copy my Proxy URL") log(" → paste that URL into the TutorialMaker Space's 'Your proxy URL' field, then Generate.") log("Tip: 'Test proxy' egress IP should be your HOME IP, and YouTube should be reachable.") log("(bore auto-downloads on first use; 'Set on Space (owner)' is only for the Space owner.)") poll() root.mainloop() if __name__ == "__main__": # Frozen (PyInstaller) re-entry: act as `python -m proxy` when asked, so the packaged # executable can run the forward proxy as a normal killable subprocess of itself. if "--run-proxy" in sys.argv: _i = sys.argv.index("--run-proxy") import proxy sys.argv = ["proxy", *sys.argv[_i + 1:]] proxy.main() else: main()