fix(youtube): address Greptile review feedback
Three changes from automated review on PR #376: 1. Add `--` option terminator before host in _wrap_ytdlp_cmd (P1 security) Prevents SSH option injection if LAST30DAYS_YT_SSH_HOST were ever set to a value starting with `-` (e.g. `-oProxyCommand=...`). Low exploitability since the env var is user-controlled config — but the fix is a single arg and turns a self-harm footgun into no footgun. 2. Hoist `import shlex` to module-level (P2 style) Pure stdlib import, no reason for the deferred form. Cleaner. 3. Cache _ytdlp_ssh_host() result in fetch_transcript (P2 style) Was being called 2-3x per video; the function is cheap (env lookup + strip) so this is purely about readability. Adds test_wrap_cmd_uses_option_terminator covering the security fix explicitly with a `-oFoo=bar` host value. Updates index assertions in the two existing tests that check command shape (host is now at index 4, command string at 5, with `--` at 3).
This commit is contained in:
@@ -10,6 +10,7 @@ import json
|
|||||||
import math
|
import math
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
|
import shlex
|
||||||
import shutil
|
import shutil
|
||||||
import sys
|
import sys
|
||||||
import tempfile
|
import tempfile
|
||||||
@@ -134,13 +135,14 @@ def _wrap_ytdlp_cmd(cmd: List[str]) -> List[str]:
|
|||||||
|
|
||||||
Args are shell-quoted to survive the remote shell. Uses BatchMode=yes so
|
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.
|
a misconfigured key fails fast instead of hanging on a password prompt.
|
||||||
|
The `--` option terminator prevents an SSH option-injection if
|
||||||
|
LAST30DAYS_YT_SSH_HOST were ever set to a value starting with `-`.
|
||||||
"""
|
"""
|
||||||
host = _ytdlp_ssh_host()
|
host = _ytdlp_ssh_host()
|
||||||
if not host:
|
if not host:
|
||||||
return cmd
|
return cmd
|
||||||
import shlex
|
|
||||||
remote_cmd = " ".join(shlex.quote(a) for a in cmd)
|
remote_cmd = " ".join(shlex.quote(a) for a in cmd)
|
||||||
return ["ssh", "-o", "BatchMode=yes", host, remote_cmd]
|
return ["ssh", "-o", "BatchMode=yes", "--", host, remote_cmd]
|
||||||
|
|
||||||
|
|
||||||
def _extract_core_subject(topic: str) -> str:
|
def _extract_core_subject(topic: str) -> str:
|
||||||
@@ -520,14 +522,15 @@ def fetch_transcript(video_id: str, temp_dir: str) -> Optional[str]:
|
|||||||
# file on the remote host that we can't easily read back. Skip it and
|
# file on the remote host that we can't easily read back. Skip it and
|
||||||
# use the HTTP transcript fallback (different YouTube endpoint, less
|
# use the HTTP transcript fallback (different YouTube endpoint, less
|
||||||
# bot-walled, works fine from datacenter IPs).
|
# bot-walled, works fine from datacenter IPs).
|
||||||
use_ytdlp = is_ytdlp_installed() and not _ytdlp_ssh_host()
|
ssh_host = _ytdlp_ssh_host()
|
||||||
|
use_ytdlp = is_ytdlp_installed() and not ssh_host
|
||||||
if use_ytdlp:
|
if use_ytdlp:
|
||||||
raw_vtt = _fetch_transcript_ytdlp(video_id, temp_dir)
|
raw_vtt = _fetch_transcript_ytdlp(video_id, temp_dir)
|
||||||
if not raw_vtt:
|
if not raw_vtt:
|
||||||
_log(f"yt-dlp transcript failed for {video_id}, trying direct HTTP fallback")
|
_log(f"yt-dlp transcript failed for {video_id}, trying direct HTTP fallback")
|
||||||
raw_vtt = _fetch_transcript_direct(video_id)
|
raw_vtt = _fetch_transcript_direct(video_id)
|
||||||
else:
|
else:
|
||||||
if _ytdlp_ssh_host():
|
if ssh_host:
|
||||||
_log("SSH-routing active, using direct HTTP transcript fetch")
|
_log("SSH-routing active, using direct HTTP transcript fetch")
|
||||||
else:
|
else:
|
||||||
_log("yt-dlp not installed, using direct HTTP transcript fetch")
|
_log("yt-dlp not installed, using direct HTTP transcript fetch")
|
||||||
|
|||||||
@@ -452,10 +452,13 @@ class TestYtdlpSSHRouting(unittest.TestCase):
|
|||||||
self.assertEqual(wrapped[0], "ssh")
|
self.assertEqual(wrapped[0], "ssh")
|
||||||
self.assertEqual(wrapped[1], "-o")
|
self.assertEqual(wrapped[1], "-o")
|
||||||
self.assertEqual(wrapped[2], "BatchMode=yes")
|
self.assertEqual(wrapped[2], "BatchMode=yes")
|
||||||
self.assertEqual(wrapped[3], "macmini")
|
# `--` terminates SSH option parsing so a host starting with `-`
|
||||||
|
# (e.g. `-oProxyCommand=...`) cannot be reinterpreted as a flag.
|
||||||
|
self.assertEqual(wrapped[3], "--")
|
||||||
|
self.assertEqual(wrapped[4], "macmini")
|
||||||
# Final arg is the shell-quoted command string
|
# Final arg is the shell-quoted command string
|
||||||
self.assertIn("yt-dlp", wrapped[4])
|
self.assertIn("yt-dlp", wrapped[5])
|
||||||
self.assertIn("ytsearch5:test", wrapped[4])
|
self.assertIn("ytsearch5:test", wrapped[5])
|
||||||
|
|
||||||
def test_wrap_cmd_quotes_args_with_spaces(self):
|
def test_wrap_cmd_quotes_args_with_spaces(self):
|
||||||
"""Args containing spaces or special chars are shell-quoted."""
|
"""Args containing spaces or special chars are shell-quoted."""
|
||||||
@@ -463,7 +466,21 @@ class TestYtdlpSSHRouting(unittest.TestCase):
|
|||||||
cmd = ["yt-dlp", "ytsearch5:hello world", "--dump-json"]
|
cmd = ["yt-dlp", "ytsearch5:hello world", "--dump-json"]
|
||||||
wrapped = youtube_yt._wrap_ytdlp_cmd(cmd)
|
wrapped = youtube_yt._wrap_ytdlp_cmd(cmd)
|
||||||
# shlex.quote wraps the whole arg in single quotes when it contains spaces
|
# shlex.quote wraps the whole arg in single quotes when it contains spaces
|
||||||
self.assertIn("'ytsearch5:hello world'", wrapped[4])
|
self.assertIn("'ytsearch5:hello world'", wrapped[5])
|
||||||
|
|
||||||
|
def test_wrap_cmd_uses_option_terminator(self):
|
||||||
|
"""`--` is inserted before host to prevent SSH option injection.
|
||||||
|
|
||||||
|
Without `--`, an env var like `LAST30DAYS_YT_SSH_HOST=-oProxyCommand=...`
|
||||||
|
would be parsed by ssh as an option flag. The terminator forces it
|
||||||
|
to be treated as a hostname (which will then fail clean if invalid).
|
||||||
|
"""
|
||||||
|
os.environ["LAST30DAYS_YT_SSH_HOST"] = "-oFoo=bar"
|
||||||
|
cmd = ["yt-dlp", "--version"]
|
||||||
|
wrapped = youtube_yt._wrap_ytdlp_cmd(cmd)
|
||||||
|
# Find the `--` terminator and verify the host comes immediately after
|
||||||
|
dash_idx = wrapped.index("--")
|
||||||
|
self.assertEqual(wrapped[dash_idx + 1], "-oFoo=bar")
|
||||||
|
|
||||||
def test_is_ytdlp_installed_short_circuits_with_ssh(self):
|
def test_is_ytdlp_installed_short_circuits_with_ssh(self):
|
||||||
"""is_ytdlp_installed returns True without local check when SSH routing is on."""
|
"""is_ytdlp_installed returns True without local check when SSH routing is on."""
|
||||||
@@ -489,11 +506,12 @@ class TestYtdlpSSHRouting(unittest.TestCase):
|
|||||||
youtube_yt.search_youtube("test", "2026-02-01", "2026-03-01")
|
youtube_yt.search_youtube("test", "2026-02-01", "2026-03-01")
|
||||||
cmd = run_mock.call_args.args[0]
|
cmd = run_mock.call_args.args[0]
|
||||||
self.assertEqual(cmd[0], "ssh")
|
self.assertEqual(cmd[0], "ssh")
|
||||||
self.assertEqual(cmd[3], "macmini")
|
self.assertEqual(cmd[3], "--")
|
||||||
# The shell-quoted yt-dlp invocation lives at index 4
|
self.assertEqual(cmd[4], "macmini")
|
||||||
self.assertIn("yt-dlp", cmd[4])
|
# The shell-quoted yt-dlp invocation lives at index 5
|
||||||
self.assertIn("--ignore-config", cmd[4])
|
self.assertIn("yt-dlp", cmd[5])
|
||||||
self.assertIn("--no-cookies-from-browser", cmd[4])
|
self.assertIn("--ignore-config", cmd[5])
|
||||||
|
self.assertIn("--no-cookies-from-browser", cmd[5])
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
Reference in New Issue
Block a user