Panniantong/Agent-Reach

Why it got this result

All subprocess invocations across the runtime (cli.py, v2ex.py, transcribe.py, probe.py) use subprocess.run() with an argument-vector list and no shell=True / os.system / eval anywhere in agent_reach/ (confirmed by grep: zero shell=True, os.system, or eval( hits in application code). Arguments passed to yt-dlp/ffmpeg/curl are either fixed flags or a URL that is first validated by _assert_safe_public_url()/normalize_public_http_url() (reject non-http(s) schemes, private/loopback/link-local/reserved IPs including inet_aton shorthand forms like 127.1/0x7f000001, and internal hostnames). The '--' separator before the URL argument in yt-dlp invocation prevents option-injection via a URL starting with '-'. Bandit flagged B603 (subprocess without shell equals true) across 25 call sites in cli.py/v2ex.py/transcribe.py -- all manually confirmed to use list-form argv with a fixed executable located via shutil.which(), no shell re-parsing.

Why it matters

Unsanitized command construction in shipped runtime code can give a remote attacker arbitrary code execution with the application's privileges.

Evidence

def _assert_safe_public_url(url: str) -> None:
    ...
    if host in _BLOCKED_HOSTS or host.endswith(".localhost"):
        raise TranscribeError("SSRF blocked: internal host is not allowed")
    if _is_private_ip(host):
        raise TranscribeError("SSRF blocked: private/internal IP is not allowed")


def download_audio(url: str, out_dir: Path) -> Path:
    """Download audio with yt-dlp into out_dir; return the resulting file path."""
    _assert_safe_public_url(url)
    _require("yt-dlp")
    template = out_dir / "source.%(ext)s"
    _run(
…

    command = [
        curl, "--fail", "--silent", "--show-error", "--proto", "=https",
        "--connect-timeout", "5", "--max-time", str(_TIMEOUT),
        "--max-filesize", str(_MAX_RESPONSE_BYTES),
        "--header", f"User-Agent: {_UA}", "--url", url,
    ]
    try:
        result = subprocess.run(
            command, capture_output=True, encoding="utf-8", errors="replace",
            timeout=_TIMEOUT + 2, env=utf8_subprocess_env(),
        )

Why it got this result

Agent Reach's declared purpose (README.md, SKILL.md, pyproject.toml description) is to give an AI agent read access to public internet platforms (Twitter, Reddit, YouTube, GitHub, etc.) for research/search -- a benign, clearly non-offensive capability. There is no malware/exploit/credential-theft/evasion generation capability anywhere in the instruction bodies or bundled scripts; the only CDP browser-automation pattern (boss.py) exists to reuse the USER's OWN already-logged-in browser session for legitimate job-search automation and explicitly requires the user to visually confirm their own logged-in state before any action -- the opposite of covert/unattributed operation. No offensive-security framing, dual-use pentest claim, or log-suppression/anti-analysis instructions are present anywhere in the repo.

Why it matters

An artifact whose purpose is to weaponize the agent -- to produce malware, steal credentials, evade defenses, or run attacks at scale -- is dangerous regardless of how securely it is built; posture controls (auth, input validation, least privilege) cannot make a weapon safe, which is why a confirmed weaponization finding caps the trust verdict to `blocked` rather than being averaged against a good posture score. Catching malicious PURPOSE is a distinct safety control, and it must be conservative because offensive capability is dual-use: legitimate pentest, CTF, defensive, and research work must abstain (UNKNOWN), never block -- unless the artifact itself hides from the people who install, run or audit it.

Evidence

description = "Give your AI Agent eyes to see the entire internet. Search + Read 10+ platforms."

只绑定 `127.0.0.1:9222` 的专用 Chrome；**拉起后第一步是暂停并让用户肉眼确认**
窗口内是已登录状态（右上角有头像），未登录则让用户登录/扫码，用户确认后再运行
`boss --cdp-url http://localhost:9222 login --cdp` 和 `agent-reach doctor` 验收。
不要让用户自己研究端口参数。

Why it got this result

The only arbitrary-URL fetch surface is WebChannel.read() (Jina Reader proxy for 'any URL'). It passes every URL through normalize_public_http_url() before use, which rejects non-HTTP(S) schemes, userinfo, literal non-global IPs (RFC1918/loopback/link-local/metadata via ipaddress.is_global), and known internal TLD/hostname suffixes (.local, .internal, metadata.google.internal, etc.). All other channels (V2EX, Xueqiu, Boss, GitHub, Bilibili) call a fixed, hardcoded API host and validate scheme/host/port with host_matches()/_validate_api_url() before any request, so user input cannot redirect the destination host. Responses are size-capped (_MAX_RESPONSE_BYTES) and timeouts are set on every urlopen/subprocess call.

Why it matters

Enables internal pivoting and data exfiltration.

Evidence

def normalize_public_http_url(url: str) -> str:
    ...
    literal_address = _literal_ip_address(host)
    if (
        parsed.scheme.lower() not in {"http", "https"}
        or not host
        or parsed.username is not None
        or parsed.password is not None
        or "%" in host
        or host in _BLOCKED_PUBLIC_FETCH_HOSTS
        or host.endswith(_BLOCKED_PUBLIC_FETCH_SUFFIXES)
        or ("." not in host and literal_address is None)
        or (literal_address is not None and not literal_address.is_global)
    ):
…

def read(self, url: str) -> str:
    url = normalize_public_http_url(url)
    jina_url = f"https://r.jina.ai/{url}"
    req = urllib.request.Request(
        jina_url,
        headers={"User-Agent": _UA, "Accept": "text/plain"},
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        body = resp.read(_MAX_RESPONSE_BYTES + 1)

def _validate_api_url(url: str) -> None:
    """Allow only the public V2EX HTTPS JSON API."""
    ...
    if (
        parsed.scheme.lower() != "https"
        or (parsed.hostname or "").lower() not in {"v2ex.com", "www.v2ex.com"}
        or port not in {None, 443}
        or parsed.username is not None
        or parsed.password is not None
        or not parsed.path.startswith("/api/")
    ):
        raise ValueError("only the V2EX HTTPS API is allowed")

Why it got this result

.env.example ships only commented-out placeholder values with no real secret material. gitleaks scan found 0 findings repo-wide. Secrets are read from ~/.agent-reach/config.yaml (created at runtime, not committed) or from process.env. Config.to_dict() masks sensitive fields as '[REDACTED]'. Semgrep's atr-c0002 credential-to-log-sink rule flagged 4 lines (cli.py:1519,1532,1536 and cookie_extract.py:401) -- all 4 were read: cli.py:1519/1532/1536 write the literal status string 'requires login' into config.set()/print() (not a credential value), and cookie_extract.py:401 writes AUTH_TOKEN/CT0 to a private 0600 file (~/.config/bird/credentials.env) via atomic_write_private_text, shell-quoted with shlex.quote -- this is a private credential file write, not a log/stdout sink, and the file is owner-only. No credential value is printed, logged, or committed anywhere.

Why it matters

Plaintext secrets are easily exfiltrated.

Evidence

        if auth_token and ct0:
            config.set("twitter_auth_token", auth_token)
            config.set("twitter_ct0", ct0)

            print("✅ Twitter cookies 已保存到 ~/.agent-reach/config.yaml")

                legacy_results = (
                    (
                        "~/.config/xfetch/session.json",
                        _sync_xfetch_session(auth_token, ct0),
                    ),
                    (
                        "~/.config/bird/credentials.env",
                        _sync_bird_env(auth_token, ct0),
                    ),
                )

        bird_dir = home_dir() / ".config" / "bird"
        make_private_dir(bird_dir)
        env_path = bird_dir / "credentials.env"
        atomic_write_private_text(
            env_path,
            f"AUTH_TOKEN={shlex.quote(auth_token)}\n"
            f"CT0={shlex.quote(ct0)}\n",
        )
        return True

+6 more locations in the finding JSON

Why it got this result

The single MCP tool (get_status) takes no arguments (empty inputSchema), eliminating that attack surface entirely. The CLI entry points use argparse with explicit choices=[...] allowlists for all enum-like parameters (configure key, --from-browser browser, --platform, format platform) so out-of-range values are rejected by argparse itself before reaching application code. Free-text inputs that matter for security (URLs passed to web/transcribe channels) are validated by dedicated schema-like validators -- normalize_public_http_url(), _validate_api_url(), _assert_safe_public_url() -- which enforce scheme/host/port/userinfo constraints before any network or subprocess call. Config file inputs are size-capped (_MAX_CONFIG_BYTES) and type-checked (must parse to a dict) before use.

Why it matters

Frontline control to prevent multiple classes of vulns.

Evidence

    p_conf = sub.add_parser("configure", help="Set a config value or auto-extract from browser")
    p_conf.add_argument("key", nargs="?", default=None,
                        choices=["proxy", "github-token", "groq-key", "openai-key",
                                 "twitter-cookies", "youtube-cookies",
                                 "xhs-cookies"],
                        help="What to configure (omit if using --from-browser)")
    p_conf.add_argument("value", nargs="*", help="The value(s) to set")

    @server.list_tools()
    async def list_tools():
        return [
            Tool(name="get_status",
                 description="Get Agent Reach status: which channels are installed and active.",
                 inputSchema={"type": "object", "properties": {}}),
        ]

        try:
            payload = read_small_text_no_follow(
                self.config_path,
                max_bytes=_MAX_CONFIG_BYTES,
            )
        except PrivatePathError as exc:
            raise ConfigSecurityError(str(exc)) from exc
        if payload is None:
            self.data = {}
            return

        loaded = yaml.safe_load(payload) or {}
        if not isinstance(loaded, dict):
            raise ConfigError("配置文件顶层必须是对象")

Why it got this result

No use of pickle, marshal, or yaml.load/yaml.unsafe_load anywhere in the codebase (grep for these patterns across agent_reach/ returned zero matches). All YAML parsing uses yaml.safe_load (config.py load(), github.py hosts.yml parsing) and all JSON parsing uses the standard json module, neither of which executes arbitrary code on untrusted input.

Why it matters

Can lead to arbitrary code execution.

Evidence

        loaded = yaml.safe_load(payload) or {}
        if not isinstance(loaded, dict):
            raise ConfigError("配置文件顶层必须是对象")
        self.data = loaded

    try:
        payload = yaml.safe_load(raw)
    except yaml.YAMLError as exc:
        raise GitHubConfigError("gh hosts.yml 不是有效的 UTF-8 YAML") from exc

Why it got this result

The only MCP tool exposed (get_status in integrations/mcp_server.py) takes no filesystem arguments at all -- it has an empty inputSchema and calls eyes.doctor_report() with no caller-supplied path. The CLI's filesystem writes (skill install, config, cookie sync) write only to fixed, hardcoded locations under the user's home directory (~/.agent-reach, ~/.claude/skills, ~/.config/bird, etc.) -- none of these paths are derived from untrusted/caller-supplied input. Every private-file read/write path goes through utils/paths.py helpers (ensure_no_symlink_path, read_small_text_no_follow, atomic_write_private_text) which walk every path component with os.lstat and reject any symlink before opening, preventing symlink-redirection attacks even though the destination paths are already fixed.

Why it matters

Traversal exposes arbitrary host files; can lead to code execution.

Evidence

    @server.list_tools()
    async def list_tools():
        return [
            Tool(name="get_status",
                 description="Get Agent Reach status: which channels are installed and active.",
                 inputSchema={"type": "object", "properties": {}}),
        ]

    @server.call_tool()
    async def call_tool(name: str, arguments: dict):
        try:
            if name == "get_status":
                result = eyes.doctor_report()

def ensure_no_symlink_path(path: str | Path, label: str = "路径") -> Path:
    """Reject any existing symlink component without resolving the path."""
    target = Path(path)
    absolute = Path(os.path.abspath(os.fspath(target)))
    current = Path(absolute.anchor)
    for part in absolute.parts[1:]:
        current /= part
        try:
            mode = os.lstat(current).st_mode
        except FileNotFoundError:
            continue
        if stat.S_ISLNK(mode):

Why it got this result

The repository has a single GitHub Actions workflow (.github/workflows/pytest.yml) triggered only by push and pull_request (not pull_request_target, issue_comment, workflow_run, or repository_dispatch). No step interpolates any ${{ github.event.* }}, ${{ inputs.* }}, or other attacker-controlled expression into a run: shell block -- all run steps execute fixed commands (pip install, pytest, python -m build, wheel verification script). The matrix python-version values are static literals defined in the workflow file itself, not derived from any external input. Semgrep's github-actions-mutable-action-tag rule flagged 6 lines (all uses: actions/checkout@v4 / actions/setup-python@v5), which is a supply-chain pinning hygiene note (mutable major-version tag vs. pinned SHA), not a script-injection finding -- there is no shell-injection vector in this workflow.

Why it matters

CI/CD script injection can execute attacker-controlled commands with workflow tokens, repository secrets, trusted-runner access, artifact or cache authority, and release or deployment privileges.

Evidence

name: ci

on:
  push:
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false
      matrix:
        python-version: ["3.10", "3.11", "3.12", "3.13"]
    steps:
…

Why it got this result

Not every HTTP response is byte-capped: several fixed-host health/status checks (bilibili.py, boss.py, xueqiu.py) use a bare resp.read() with no explicit cap. However, these calls target fixed, hardcoded API hosts rather than an attacker-selectable destination. The paths that are both attacker-triggerable and high-cost -- media download (yt-dlp, MAX_SOURCE_BYTES/MAX_AUDIO_SECONDS/MAX_CHUNKS) and the arbitrary-URL web-reader fetch (_MAX_RESPONSE_BYTES) -- remain explicitly bounded, and every urlopen/subprocess call carries a timeout. The exposed MCP surface itself is a single fixed, argument-free status tool with no concurrent untrusted request handling, so resource-exhaustion protection rests on bounding the attacker-controllable paths rather than on a universal response-size cap.

Why it matters

Resource exhaustion can crash or stall MCP.

Evidence

SIZE_LIMIT_BYTES = 24 * 1024 * 1024
CHUNK_SECONDS = 600
MAX_SOURCE_BYTES = 512 * 1024 * 1024
MAX_CHUNKS = 24
MAX_TOTAL_CHUNK_BYTES = 96 * 1024 * 1024
MAX_AUDIO_SECONDS = MAX_CHUNKS * CHUNK_SECONDS
FFPROBE_TIMEOUT_SECONDS = 30

_MAX_RESPONSE_BYTES = 5 * 1024 * 1024
_ANTIBOT_SCAN_BYTES = 4096

        with urllib.request.urlopen(req, timeout=30) as resp:
            body = resp.read(_MAX_RESPONSE_BYTES + 1)
        if len(body) > _MAX_RESPONSE_BYTES:
            raise ValueError(
                f"Jina Reader response exceeds {_MAX_RESPONSE_BYTES} byte limit"
            )

Why it got this result

The only billable external dependency is the Whisper transcription API (Groq/OpenAI), gated behind a user-supplied API key the user must configure themselves (no key ships with the repo). Each transcription run is bounded: MAX_SOURCE_BYTES (512MB download cap), MAX_AUDIO_SECONDS (4 hours via MAX_CHUNKS=24 * CHUNK_SECONDS=600) reject oversized/too-long media before any paid API call, and chunks are capped at MAX_TOTAL_CHUNK_BYTES. There is no loop or batch mode that could multiply billed calls unboundedly; transcribe() processes one source per invocation. All other channels (web/social/search) either use free public APIs or require the operator's own paid key (Exa), so Agent Reach itself has no mechanism to run up a bill beyond what the operator explicitly triggers per call.

Why it matters

Prevents cost-based disruption.

Evidence

MAX_SOURCE_BYTES = 512 * 1024 * 1024
MAX_CHUNKS = 24  # 4 hours at the standard 10-minute segment size
MAX_TOTAL_CHUNK_BYTES = 96 * 1024 * 1024
MAX_AUDIO_SECONDS = MAX_CHUNKS * CHUNK_SECONDS

def _require_duration_within_budget(path: Path) -> float:
    """Reject audio that cannot fit within the bounded chunk budget."""
    duration = _probe_audio_duration(path)
    if duration > MAX_AUDIO_SECONDS:
        max_minutes = MAX_AUDIO_SECONDS // 60
        raise TranscribeError(
            f"audio duration exceeds safety limit of {max_minutes} minutes"
        )
    return duration

Why it got this result

Agent Reach ships as a Python package (pip-installable CLI/MCP tool) with no Dockerfile, docker-compose, Helm chart, or Kubernetes manifest anywhere in the repository (confirmed: find for Dockerfile*/docker-compose*/*.tf/*.yaml under deploy-style paths returned nothing beyond the CI workflow and mcporter config). There is no container or cloud deployment artifact for this rule's container-hardening controls (USER, read-only FS, capabilities, seccomp, hostNetwork) to apply to.

Why it matters

Reduces blast radius of exploits.

Evidence

No Dockerfile, docker-compose.yml, Helm chart, or Kubernetes/Terraform manifest found in the repository tree (verified via recursive find for Dockerfile*, docker-compose*, *.tf, helm/, k8s/).