feat(youtube): route yt-dlp through SSH host for residential IP egress

Adds LAST30DAYS_YT_SSH_HOST env var (or `~/.config/last30days/.env` key).
When set, yt-dlp YouTube search invocations are wrapped as
`ssh <host> "yt-dlp ..."` so they run on a residential-IP machine.

Motivation: when last30days runs on a datacenter VPS (Hetzner,
DigitalOcean, AWS, etc.), `ytsearch:` queries return 0 results because
YouTube's bot-wall fingerprints datacenter IP ranges before any cookie
check runs. Cookies alone don't fix this — the IP reputation is checked
first. Verified across yt-dlp stable 2026.03.17 and nightly builds.

The existing fallbacks (browser cookies, residential proxy services,
excluding YouTube) all have downsides: cookies expire, proxies cost
money, exclusion loses signal. Many users with a Mac mini, Pi, or
home server can host yt-dlp on their own residential IP — this just
needs an SSH alias and a one-line env var to wire it up.

Behaviour:
- Default (env var unset): identical to before, no shape change.
- Env var set: search command list is wrapped with `ssh -o BatchMode=yes
  <host> "<shell-quoted yt-dlp invocation>"`. is_ytdlp_installed()
  returns True without a local PATH check (the binary lives on the
  remote host).
- Transcript path: when SSH-routing is on, skips the yt-dlp transcript
  path (which writes a VTT file we couldn't easily read back over SSH)
  and uses the existing _fetch_transcript_direct HTTP fallback. The
  timedtext API isn't bot-walled, so this works fine on datacenter IPs.

Setup pitfall documented in the function docstring: on macOS hosts
with Homebrew, `eval "$(/opt/homebrew/bin/brew shellenv zsh)"` must
live in ~/.zshenv (not just ~/.zprofile) — non-login SSH shells don't
source .zprofile, so without this `ssh macmini "yt-dlp ..."` returns
"command not found" while interactive SSH works fine.

Tests: 10 new cases covering env var read, whitespace stripping,
empty-value handling, command wrapping passthrough/active modes,
shlex quoting, is_ytdlp_installed short-circuit, and end-to-end
search_youtube wrapping. Full test suite: 0 new failures (the 14
pre-existing failures in test_store, test_watchlist, test_setup_openclaw,
test_safari_cookies, test_version_consistency are unchanged on main).

Verified live: 0 results → 4 real hits for "claude code" search from a
Hetzner VPS routed through a Mac mini exit node on Tailscale.
This commit is contained in:
shoobee
2026-05-10 20:29:24 +00:00
committed by Trevin Chow
parent 602de1ebda
commit 79b5d049ce
5 changed files with 156 additions and 3 deletions
+4
View File
@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- `LAST30DAYS_YT_SSH_HOST` env var: when set, yt-dlp YouTube search invocations are routed through `ssh <host>` for residential-IP egress. Bypasses YouTube's bot-wall on datacenter IPs (Hetzner/DigitalOcean/AWS) where `ytsearch:` returns 0 results regardless of cookies (the IP fingerprint is checked first). The named host must be configured in `~/.ssh/config` and have yt-dlp installed. The transcript path is unchanged (uses the existing HTTP fallback when SSH-routing is on, since the timedtext API isn't bot-walled).
### Changed
- Replace the SKILL_ROOT resolver loops in Step 1 and comparison-mode with a single `SKILL_DIR` substitution pattern. The model templates the absolute path of the SKILL.md's own directory (which it always knows from the Read tool result); the bash block just validates that `scripts/last30days.py` lives there. Removes ~80 lines of bash across the two locations. Fixes a real bug: the previous resolver could pick a different install than the SKILL.md the model loaded from (spec-vs-engine divergence) and didn't enumerate harnesses like Hermes at all. The simplification works for any harness without enumeration because it just uses wherever SKILL.md was loaded from. STEP 0's marketplaces-stale-clone hop is unchanged.
+7
View File
@@ -541,6 +541,13 @@ def main() -> int:
config = env.get_config()
# Surface SSH-routing config as an env var so library modules (e.g.
# youtube_yt) can read it without taking a config dependency. This
# routes yt-dlp through `ssh <host>` to bypass YouTube's bot-wall on
# datacenter IPs (see lib/youtube_yt.py for details).
if config.get("LAST30DAYS_YT_SSH_HOST") and "LAST30DAYS_YT_SSH_HOST" not in os.environ:
os.environ["LAST30DAYS_YT_SSH_HOST"] = config["LAST30DAYS_YT_SSH_HOST"]
# Handle setup subcommand
topic = " ".join(args.topic).strip()
if topic.lower() == "setup":
+1
View File
@@ -329,6 +329,7 @@ def get_config() -> dict[str, Any]:
('SETUP_COMPLETE', None),
('INCLUDE_SOURCES', ''),
('EXCLUDE_SOURCES', ''),
('LAST30DAYS_YOUTUBE_SSH_HOST', None),
]
for key, default in keys:
+55 -3
View File
@@ -8,6 +8,7 @@ Inspired by Peter Steinberger's toolchain approach (yt-dlp + summarize CLI).
import json
import math
import os
import re
import shutil
import sys
@@ -96,10 +97,52 @@ def _log(msg: str):
def is_ytdlp_installed() -> bool:
"""Check if yt-dlp is available in PATH."""
"""Check if yt-dlp is available locally, or if SSH routing is configured.
When LAST30DAYS_YT_SSH_HOST is set, returns True without a local check —
yt-dlp lives on the remote host. Failures surface naturally on first use.
"""
if _ytdlp_ssh_host():
return True
return shutil.which("yt-dlp") is not None
def _ytdlp_ssh_host() -> Optional[str]:
"""Return SSH host alias if yt-dlp should be routed via SSH, else None.
Set LAST30DAYS_YT_SSH_HOST=<ssh-alias> (e.g. 'macmini') in the environment
to route yt-dlp through SSH for residential IP egress. This bypasses
YouTube's bot-wall on datacenter IPs (Hetzner, DigitalOcean, AWS, etc.)
where ytsearch returns 0 results regardless of cookies.
The remote host must have yt-dlp installed and reachable via the named
SSH alias (configured in ~/.ssh/config). On macOS hosts with Homebrew,
add brew shellenv to ~/.zshenv (not just ~/.zprofile) so non-login SSH
shells find yt-dlp on PATH.
To use a value from ~/.config/last30days/.env, export it into the
environment before invoking the engine, e.g. in a wrapper:
set -a; source ~/.config/last30days/.env; set +a
python3 last30days.py "..."
"""
host = os.environ.get("LAST30DAYS_YT_SSH_HOST", "").strip()
return host or None
def _wrap_ytdlp_cmd(cmd: List[str]) -> List[str]:
"""Wrap a yt-dlp command list with `ssh <host>` when SSH routing is set.
Args are shell-quoted to survive the remote shell. Uses BatchMode=yes so
a misconfigured key fails fast instead of hanging on a password prompt.
"""
host = _ytdlp_ssh_host()
if not host:
return cmd
import shlex
remote_cmd = " ".join(shlex.quote(a) for a in cmd)
return ["ssh", "-o", "BatchMode=yes", host, remote_cmd]
def _extract_core_subject(topic: str) -> str:
"""Extract core subject from verbose query for YouTube search.
@@ -223,6 +266,7 @@ def search_youtube(
"--no-warnings",
"--no-download",
]
cmd = _wrap_ytdlp_cmd(cmd)
try:
result = subproc.run_with_timeout(cmd, timeout=120)
@@ -472,13 +516,21 @@ def fetch_transcript(video_id: str, temp_dir: str) -> Optional[str]:
Plaintext transcript string, or None if no captions available.
"""
raw_vtt = None
if is_ytdlp_installed():
# When SSH-routing is on, the yt-dlp transcript path would write a VTT
# file on the remote host that we can't easily read back. Skip it and
# use the HTTP transcript fallback (different YouTube endpoint, less
# bot-walled, works fine from datacenter IPs).
use_ytdlp = is_ytdlp_installed() and not _ytdlp_ssh_host()
if use_ytdlp:
raw_vtt = _fetch_transcript_ytdlp(video_id, temp_dir)
if not raw_vtt:
_log(f"yt-dlp transcript failed for {video_id}, trying direct HTTP fallback")
raw_vtt = _fetch_transcript_direct(video_id)
else:
_log("yt-dlp not installed, using direct HTTP transcript fetch")
if _ytdlp_ssh_host():
_log("SSH-routing active, using direct HTTP transcript fetch")
else:
_log("yt-dlp not installed, using direct HTTP transcript fetch")
raw_vtt = _fetch_transcript_direct(video_id)
if not raw_vtt:
+89
View File
@@ -1,6 +1,7 @@
"""Tests for YouTube transcript highlights and yt-dlp safety flags."""
import json
import os
import sys
import tempfile
import unittest
@@ -407,5 +408,93 @@ class TestSearchAndTranscribe(unittest.TestCase):
ft_mock.assert_not_called()
class TestYtdlpSSHRouting(unittest.TestCase):
"""LAST30DAYS_YT_SSH_HOST routes yt-dlp invocations through SSH for residential IP."""
def setUp(self):
# Ensure clean env for each test
self._saved_env = os.environ.pop("LAST30DAYS_YT_SSH_HOST", None)
def tearDown(self):
os.environ.pop("LAST30DAYS_YT_SSH_HOST", None)
if self._saved_env is not None:
os.environ["LAST30DAYS_YT_SSH_HOST"] = self._saved_env
def test_no_env_var_returns_none(self):
"""Without the env var set, _ytdlp_ssh_host returns None."""
self.assertIsNone(youtube_yt._ytdlp_ssh_host())
def test_env_var_returns_host(self):
"""With LAST30DAYS_YT_SSH_HOST set, _ytdlp_ssh_host returns it."""
os.environ["LAST30DAYS_YT_SSH_HOST"] = "macmini"
self.assertEqual(youtube_yt._ytdlp_ssh_host(), "macmini")
def test_env_var_whitespace_stripped(self):
"""Whitespace around the host alias is stripped."""
os.environ["LAST30DAYS_YT_SSH_HOST"] = " macmini "
self.assertEqual(youtube_yt._ytdlp_ssh_host(), "macmini")
def test_empty_env_var_falls_back_to_none(self):
"""An empty env var is treated as unset."""
os.environ["LAST30DAYS_YT_SSH_HOST"] = ""
self.assertIsNone(youtube_yt._ytdlp_ssh_host())
def test_wrap_cmd_passthrough_when_unset(self):
"""_wrap_ytdlp_cmd returns input unchanged when SSH routing is off."""
cmd = ["yt-dlp", "--ignore-config", "ytsearch5:test"]
self.assertEqual(youtube_yt._wrap_ytdlp_cmd(cmd), cmd)
def test_wrap_cmd_prepends_ssh_when_set(self):
"""_wrap_ytdlp_cmd prepends ssh <host> when SSH routing is on."""
os.environ["LAST30DAYS_YT_SSH_HOST"] = "macmini"
cmd = ["yt-dlp", "--ignore-config", "ytsearch5:test"]
wrapped = youtube_yt._wrap_ytdlp_cmd(cmd)
self.assertEqual(wrapped[0], "ssh")
self.assertEqual(wrapped[1], "-o")
self.assertEqual(wrapped[2], "BatchMode=yes")
self.assertEqual(wrapped[3], "macmini")
# Final arg is the shell-quoted command string
self.assertIn("yt-dlp", wrapped[4])
self.assertIn("ytsearch5:test", wrapped[4])
def test_wrap_cmd_quotes_args_with_spaces(self):
"""Args containing spaces or special chars are shell-quoted."""
os.environ["LAST30DAYS_YT_SSH_HOST"] = "macmini"
cmd = ["yt-dlp", "ytsearch5:hello world", "--dump-json"]
wrapped = youtube_yt._wrap_ytdlp_cmd(cmd)
# shlex.quote wraps the whole arg in single quotes when it contains spaces
self.assertIn("'ytsearch5:hello world'", wrapped[4])
def test_is_ytdlp_installed_short_circuits_with_ssh(self):
"""is_ytdlp_installed returns True without local check when SSH routing is on."""
os.environ["LAST30DAYS_YT_SSH_HOST"] = "macmini"
with mock.patch("lib.youtube_yt.shutil.which", return_value=None) as which_mock:
self.assertTrue(youtube_yt.is_ytdlp_installed())
which_mock.assert_not_called()
def test_is_ytdlp_installed_falls_through_without_ssh(self):
"""is_ytdlp_installed checks PATH normally when SSH routing is off."""
with mock.patch("lib.youtube_yt.shutil.which", return_value="/usr/bin/yt-dlp"):
self.assertTrue(youtube_yt.is_ytdlp_installed())
with mock.patch("lib.youtube_yt.shutil.which", return_value=None):
self.assertFalse(youtube_yt.is_ytdlp_installed())
def test_search_call_routes_through_ssh(self):
"""search_youtube wraps the yt-dlp invocation when SSH routing is on."""
os.environ["LAST30DAYS_YT_SSH_HOST"] = "macmini"
from lib.subproc import SubprocResult
fake_result = SubprocResult(returncode=0, stdout="", stderr="")
with mock.patch.object(youtube_yt.subproc, "run_with_timeout",
return_value=fake_result) as run_mock:
youtube_yt.search_youtube("test", "2026-02-01", "2026-03-01")
cmd = run_mock.call_args.args[0]
self.assertEqual(cmd[0], "ssh")
self.assertEqual(cmd[3], "macmini")
# The shell-quoted yt-dlp invocation lives at index 4
self.assertIn("yt-dlp", cmd[4])
self.assertIn("--ignore-config", cmd[4])
self.assertIn("--no-cookies-from-browser", cmd[4])
if __name__ == "__main__":
unittest.main()