security: fix yt-dlp arg-injection RCE, SSRF, and install-time bash; mask session cookies

.github/workflows/pytest.yml

@@ -4,6 +4,10 @@ on:
   push:
   pull_request:
 
+# Least privilege: the workflow only needs to read the repo to run tests.
+permissions:
+  contents: read
+
 jobs:
   test:
     runs-on: ubuntu-latest

agent_reach/channels/web.py

@@ -1,10 +1,16 @@
 # -*- coding: utf-8 -*-
 """Web — any URL via Jina Reader. Always available."""
 
+import urllib.parse
 import urllib.request
+
+from ..utils.urlsafe import assert_safe_public_url, is_http_url
 from .base import Channel
 
 _UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
+# Cap the Jina Reader response so a hostile/huge page can't exhaust memory or
+# flood the agent's context (and token budget).
+_MAX_BYTES = 10 * 1024 * 1024  # 10 MB
 
 
 class WebChannel(Channel):
@@ -23,12 +29,19 @@ def check(self, config=None):
 
     def read(self, url: str) -> str:
         """通过 Jina Reader 读取网页，返回 Markdown 全文。"""
-        if not url.startswith(("http://", "https://")):
+        if not is_http_url(url):
             url = "https://" + url
-        jina_url = f"https://r.jina.ai/{url}"
+        # `url` is user/agent supplied — reject non-http(s) schemes and internal
+        # SSRF targets, and percent-encode before embedding it in the Jina path
+        # so it cannot inject extra path/query segments or CRLF.
+        assert_safe_public_url(url)
+        jina_url = "https://r.jina.ai/" + urllib.parse.quote(url, safe=":/?#[]@!$&'()*+,;=~-._")
         req = urllib.request.Request(
             jina_url,
             headers={"User-Agent": _UA, "Accept": "text/plain"},
         )
         with urllib.request.urlopen(req, timeout=30) as resp:
-            return resp.read().decode("utf-8")
+            data = resp.read(_MAX_BYTES + 1)
+        if len(data) > _MAX_BYTES:
+            data = data[:_MAX_BYTES]
+        return data.decode("utf-8", errors="replace")

agent_reach/cli.py

@@ -567,10 +567,23 @@ def _install_system_deps():
     # ── Node.js (needed for mcporter) ──
     if shutil.which("node") and shutil.which("npm"):
         print("  ✅ Node.js already installed")
+    elif os.environ.get("AGENT_REACH_ALLOW_REMOTE_SCRIPTS") != "1":
+        # Downloading a remote setup script and running it through `bash` is a
+        # code-execution-on-install vector (a MITM or NodeSource compromise =
+        # arbitrary code on this host). Disabled by default; opt in explicitly.
+        print("  [!]  Node.js not found — skipping automatic NodeSource setup script.")
+        print("       (Running a remote script through bash is disabled by default.)")
+        print("       Install Node.js with a trusted method instead:")
+        print("         • nvm install 22            (recommended)")
+        print("         • https://nodejs.org        (official installer)")
+        print("         • apt install nodejs npm    (distro packages)")
+        print("       …or opt in to the NodeSource script for this run:")
+        print("         AGENT_REACH_ALLOW_REMOTE_SCRIPTS=1 agent-reach install ...")
     else:
-        print("  Installing Node.js...")
+        print("  Installing Node.js via NodeSource (AGENT_REACH_ALLOW_REMOTE_SCRIPTS=1)...")
         try:
-            # Use NodeSource setup script without invoking a shell pipeline.
+            # Opt-in only. Fetch over TLS (curl -fsSL fails closed on TLS error)
+            # to a temp file, then run via bash — no shell pipeline.
             with tempfile.NamedTemporaryFile(delete=False, suffix=".sh") as tf:
                 script_path = tf.name
             subprocess.run(

agent_reach/config.py

@@ -101,9 +101,17 @@ def get_configured_features(self) -> dict:
 
     def to_dict(self) -> dict:
         """Return config as dict (masks sensitive values)."""
+        # Mask anything credential-bearing. Session cookies (xhs_cookie,
+        # xueqiu_cookie), Bilibili SESSDATA/csrf, and saved auth/secrets do not
+        # contain "token"/"key", so they must be matched explicitly or they leak
+        # in plaintext through any diagnostic that prints to_dict().
+        sensitive = (
+            "key", "token", "password", "proxy",
+            "cookie", "secret", "session", "sessdata", "csrf", "auth", "cred",
+        )
         masked = {}
         for k, v in self.data.items():
-            if any(s in k.lower() for s in ("key", "token", "password", "proxy")):
+            if any(s in k.lower() for s in sensitive):
                 masked[k] = f"{str(v)[:8]}..." if v else None
             else:
                 masked[k] = v

agent_reach/transcribe.py

@@ -22,6 +22,7 @@
 import requests
 
 from agent_reach.config import Config
+from agent_reach.utils.urlsafe import UnsafeURLError, assert_safe_public_url
 
 # Whisper API limit is 25MB; leave headroom for multipart overhead.
 SIZE_LIMIT_BYTES = 24 * 1024 * 1024
@@ -77,6 +78,14 @@ def _run(cmd: List[str], timeout: int = 600) -> None:
 def download_audio(url: str, out_dir: Path) -> Path:
     """Download audio with yt-dlp into out_dir; return the resulting file path."""
     _require("yt-dlp")
+    # `url` may come from an LLM or scraped content. Validate it is a public
+    # http(s) target so it cannot (a) be parsed by yt-dlp as an option such as
+    # `--exec`/`--config-locations` (argument-injection → RCE / local-file read)
+    # or (b) point at an internal/metadata host (SSRF).
+    try:
+        assert_safe_public_url(url)
+    except UnsafeURLError as e:
+        raise TranscribeError(f"refusing to download unsafe source: {e}") from e
     template = out_dir / "source.%(ext)s"
     _run(
         [
@@ -88,6 +97,7 @@ def download_audio(url: str, out_dir: Path) -> Path:
             "0",
             "-o",
             str(template),
+            "--",  # end of options: `url` can never be parsed as a flag
             url,
         ],
         timeout=1800,  # long podcasts over slow networks — generous but bounded

agent_reach/utils/urlsafe.py

@@ -0,0 +1,105 @@
+# -*- coding: utf-8 -*-
+"""URL safety helpers — scheme allowlisting and SSRF guarding.
+
+Used wherever a user- or agent-supplied URL is handed to a fetcher or an
+external downloader (yt-dlp, ffmpeg, Jina Reader). The threat model: the URL
+may originate from an LLM or from scraped content, so it must not be trusted to
+point at the public web. These checks block
+
+  * non-http(s) schemes (``file:``, ``data:``, ``gopher:`` …),
+  * argument injection (a value like ``--exec=...`` has no scheme → rejected),
+  * SSRF against loopback / private / link-local hosts and the cloud metadata
+    endpoint ``169.254.169.254``,
+  * CRLF / whitespace injection when the URL is concatenated into a request.
+
+Note: ``assert_safe_public_url`` validates the *initial* host only. It does not
+constrain HTTP redirects, and resolving-then-fetching leaves a small
+DNS-rebinding window. These are pragmatic mitigations for a developer CLI, not a
+hardened SSRF proxy.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from urllib.parse import urlsplit
+
+ALLOWED_SCHEMES = ("http", "https")
+
+
+class UnsafeURLError(ValueError):
+    """Raised when a URL is not a safe public http(s) target."""
+
+
+def is_http_url(value: object) -> bool:
+    """True if ``value`` already carries an ``http://`` or ``https://`` scheme."""
+    return isinstance(value, str) and value.lower().startswith(("http://", "https://"))
+
+
+def _is_public_ip(ip: str) -> bool:
+    try:
+        addr = ipaddress.ip_address(ip)
+    except ValueError:
+        return False
+    return not (
+        addr.is_private
+        or addr.is_loopback
+        or addr.is_link_local
+        or addr.is_reserved
+        or addr.is_multicast
+        or addr.is_unspecified
+    )
+
+
+def assert_safe_public_url(url: str, *, resolve: bool = True) -> str:
+    """Validate that ``url`` is a public http(s) URL; return it unchanged.
+
+    Raises :class:`UnsafeURLError` if the scheme is not http/https, the host is
+    missing, the URL contains control/whitespace characters, or (when
+    ``resolve`` is True) the host resolves to a private/loopback/link-local/
+    reserved/metadata address.
+    """
+    if not isinstance(url, str) or not url.strip():
+        raise UnsafeURLError("empty URL")
+    # Reject control chars and whitespace early — these enable request-line /
+    # header (CRLF) injection when the URL is embedded into another request.
+    if any(ord(c) < 0x21 for c in url):
+        raise UnsafeURLError("URL contains whitespace or control characters")
+
+    parts = urlsplit(url)
+    if parts.scheme.lower() not in ALLOWED_SCHEMES:
+        raise UnsafeURLError(
+            f"unsupported URL scheme: {parts.scheme!r} (only http/https allowed)"
+        )
+    host = parts.hostname
+    if not host:
+        raise UnsafeURLError("URL has no host")
+    if not resolve:
+        return url
+
+    # Literal IP host — check directly, no DNS lookup. Determine literal-ness
+    # first; do NOT let UnsafeURLError (a ValueError subclass) be swallowed by
+    # the ip_address() parse guard.
+    try:
+        ipaddress.ip_address(host)
+        is_literal = True
+    except ValueError:
+        is_literal = False  # not a literal IP — resolve the name below
+    if is_literal:
+        if not _is_public_ip(host):
+            raise UnsafeURLError(f"URL host {host} is not a public address")
+        return url
+
+    try:
+        infos = socket.getaddrinfo(host, parts.port, proto=socket.IPPROTO_TCP)
+    except socket.gaierror as e:
+        raise UnsafeURLError(f"could not resolve host {host}: {e}") from e
+    addrs = {str(info[4][0]) for info in infos}
+    if not addrs:
+        raise UnsafeURLError(f"host {host} resolved to no addresses")
+    for ip in addrs:
+        if not _is_public_ip(ip):
+            raise UnsafeURLError(
+                f"URL host {host} resolves to non-public address {ip}"
+            )
+    return url

tests/test_config.py

@@ -74,6 +74,17 @@ def test_to_dict_masks_sensitive(self, tmp_config):
         assert masked["exa_api_key"] == "super-se..."
         assert masked["normal_setting"] == "visible"
 
+    def test_to_dict_masks_session_cookies(self, tmp_config):
+        # Session cookies / SESSDATA contain no "key"/"token" substring and
+        # must still be masked so they never leak through diagnostics.
+        tmp_config.set("xhs_cookie", "web_session=abcdef123456")
+        tmp_config.set("xueqiu_cookie", "xq_a_token=deadbeefcafe")
+        tmp_config.set("bilibili_sessdata", "SESSDATA=zzzzzzzzzzzz")
+        masked = tmp_config.to_dict()
+        assert masked["xhs_cookie"] == "web_sess..."
+        assert masked["xueqiu_cookie"] == "xq_a_tok..."
+        assert masked["bilibili_sessdata"] == "SESSDATA..."
+
     def test_save_creates_file_with_restricted_permissions(self, tmp_path):
         import stat
         import sys

tests/test_transcribe.py

@@ -225,6 +225,43 @@ def test_invalid_provider_string(self, fake_config, chunk_file):
             tr.transcribe(str(chunk_file), provider="azure", config=fake_config)
 
 
+# --- download_audio: argument-injection / SSRF guard ------------------- #
+
+
+class TestDownloadAudioSafety:
+    def test_passes_dash_dash_terminator_for_valid_url(self, monkeypatch, tmp_path):
+        captured = {}
+
+        monkeypatch.setattr(tr, "_require", lambda binary: None)
+        monkeypatch.setattr(tr, "_run", lambda cmd, timeout=600: captured.update(cmd=cmd))
+        # Avoid the empty-output check by dropping a matching file.
+        (tmp_path / "source.m4a").write_bytes(b"x")
+
+        tr.download_audio("https://1.1.1.1/video", tmp_path)
+
+        cmd = captured["cmd"]
+        # The URL must be the final arg and immediately preceded by "--" so it
+        # can never be parsed by yt-dlp as an option.
+        assert cmd[-1] == "https://1.1.1.1/video"
+        assert cmd[-2] == "--"
+
+    @pytest.mark.parametrize(
+        "bad_source",
+        ["--exec=touch pwned", "file:///etc/passwd", "http://169.254.169.254/"],
+    )
+    def test_rejects_unsafe_source(self, monkeypatch, tmp_path, bad_source):
+        ran = {"called": False}
+        monkeypatch.setattr(tr, "_require", lambda binary: None)
+
+        def boom(*a, **k):
+            ran["called"] = True
+
+        monkeypatch.setattr(tr, "_run", boom)
+        with pytest.raises(tr.TranscribeError, match="unsafe source"):
+            tr.download_audio(bad_source, tmp_path)
+        assert ran["called"] is False  # never reached yt-dlp
+
+
 # --- YouTubeChannel integration --------------------------------------- #
 
 

tests/test_urlsafe.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+"""Tests for agent_reach.utils.urlsafe — scheme allowlist + SSRF guard."""
+
+import pytest
+
+from agent_reach.utils import urlsafe as us
+
+
+class TestIsHttpUrl:
+    def test_true_for_http_https(self):
+        assert us.is_http_url("http://example.com")
+        assert us.is_http_url("https://example.com")
+        assert us.is_http_url("HTTPS://EXAMPLE.COM")
+
+    def test_false_for_bare_or_other_scheme(self):
+        assert not us.is_http_url("example.com")
+        assert not us.is_http_url("file:///etc/passwd")
+        assert not us.is_http_url("--exec=touch pwned")
+        assert not us.is_http_url(None)
+
+
+class TestAssertSafePublicURL:
+    def test_accepts_public_literal_ip(self):
+        # 1.1.1.1 is public; literal IPs skip DNS resolution.
+        assert us.assert_safe_public_url("https://1.1.1.1/path") == "https://1.1.1.1/path"
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "file:///etc/passwd",
+            "data:text/plain,hi",
+            "gopher://evil/",
+            "ftp://example.com/x",
+        ],
+    )
+    def test_rejects_non_http_scheme(self, url):
+        with pytest.raises(us.UnsafeURLError):
+            us.assert_safe_public_url(url)
+
+    @pytest.mark.parametrize(
+        "value",
+        ["--exec=touch pwned", "-J", "--config-locations=/tmp/x"],
+    )
+    def test_rejects_argument_injection_strings(self, value):
+        # yt-dlp flags have no scheme → rejected before they reach argv.
+        with pytest.raises(us.UnsafeURLError):
+            us.assert_safe_public_url(value)
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://127.0.0.1/",
+            "http://localhost/",         # resolves to loopback
+            "http://169.254.169.254/latest/meta-data/",  # cloud metadata
+            "http://10.0.0.5/",
+            "http://192.168.1.1/",
+            "http://[::1]/",
+        ],
+    )
+    def test_rejects_internal_ssrf_targets(self, url):
+        with pytest.raises(us.UnsafeURLError):
+            us.assert_safe_public_url(url)
+
+    def test_rejects_crlf_and_whitespace(self):
+        with pytest.raises(us.UnsafeURLError):
+            us.assert_safe_public_url("https://example.com/\r\nHost: evil")
+        with pytest.raises(us.UnsafeURLError):
+            us.assert_safe_public_url("https://exa mple.com/")
+
+    def test_rejects_empty(self):
+        with pytest.raises(us.UnsafeURLError):
+            us.assert_safe_public_url("")
+
+    def test_resolve_false_skips_dns(self):
+        # With resolve=False we only check scheme/host shape, not the address.
+        assert us.assert_safe_public_url("https://example.com", resolve=False)
+
+    def test_resolved_private_address_blocked(self, monkeypatch):
+        # A public-looking name that resolves to a private address is blocked.
+        monkeypatch.setattr(
+            us.socket,
+            "getaddrinfo",
+            lambda *a, **k: [(2, 1, 6, "", ("10.1.2.3", 0))],
+        )
+        with pytest.raises(us.UnsafeURLError):
+            us.assert_safe_public_url("https://sneaky.example")
+
+    def test_private_literal_ip_rejected_without_dns(self, monkeypatch):
+        # Regression: the literal-IP guard must reject on its own, not rely on
+        # getaddrinfo re-resolving the literal. Break DNS to prove it.
+        def boom(*a, **k):
+            raise AssertionError("getaddrinfo must not be called for a literal IP")
+
+        monkeypatch.setattr(us.socket, "getaddrinfo", boom)
+        with pytest.raises(us.UnsafeURLError, match="not a public address"):
+            us.assert_safe_public_url("http://10.0.0.5/")
+
+    def test_resolved_public_address_allowed(self, monkeypatch):
+        monkeypatch.setattr(
+            us.socket,
+            "getaddrinfo",
+            lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 0))],
+        )
+        assert us.assert_safe_public_url("https://example.com") == "https://example.com"