From fb3037daa5b63f598f9ba92d4eaf0ec8d2095068 Mon Sep 17 00:00:00 2001 From: wyytjh Date: Fri, 17 Jul 2026 00:36:31 +0800 Subject: [PATCH] refactor: security hardening for credential storage, input validation, and access controls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Encrypt sensitive config values (API keys, tokens, cookies) at rest using a machine-local derived key with integrity verification - Restrict Config.get() environment variable fallback to an explicit allowlist, preventing unintended leakage of unrelated env vars - Add URL validation (SSRF protection) to WebChannel.read() before proxying requests to the Jina Reader service - Add file extension validation for local audio file paths in transcribe() – only recognized audio formats are accepted - Add argument separator guard ('--') before user-supplied URL in yt-dlp subprocess call for defense-in-depth - Document cookie extraction scope in privacy notice Backward compatible: existing unencrypted configs continue to work; encrypted fields are transparently decrypted on read. All existing tests pass. --- agent_reach/channels/web.py | 82 ++++++++++++++++- agent_reach/config.py | 169 +++++++++++++++++++++++++++++++--- agent_reach/cookie_extract.py | 7 +- agent_reach/transcribe.py | 14 +++ tests/test_config.py | 10 +- 5 files changed, 260 insertions(+), 22 deletions(-) diff --git a/agent_reach/channels/web.py b/agent_reach/channels/web.py index 9d10dfe1..398c0925 100644 --- a/agent_reach/channels/web.py +++ b/agent_reach/channels/web.py @@ -1,11 +1,81 @@ # -*- coding: utf-8 -*- """Web — any URL via Jina Reader. Always available.""" +import ipaddress +import socket import urllib.request +from urllib.parse import urlparse + from .base import Channel _UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36" +#: Hostnames that are always blocked — resolved before any DNS query. +_SSRF_BLOCKED_HOSTNAMES = frozenset({ + "localhost", "127.0.0.1", "::1", "0.0.0.0", + "metadata.google.internal", "metadata.goog", +}) + +#: Cloud metadata IPs — the #1 SSRF target in cloud environments. +_SSRF_METADATA_IPS = frozenset({ + "169.254.169.254", "169.254.170.2", "169.254.169.253", + "100.100.100.200", "fd00:ec2::254", +}) + + +def _validate_url(url: str) -> None: + """Validate *url* does not target a private or internal host. + + Raises ``ValueError`` with a descriptive message when the URL is unsafe. + """ + parsed = urlparse(url) + host = (parsed.hostname or "").strip().lower() + if not host: + raise ValueError("URL has no hostname") + + # 1. Static hostname blocklist (fast, no DNS) + if host in _SSRF_BLOCKED_HOSTNAMES: + raise ValueError(f"URL target '{host}' is not allowed") + + # 2. Cloud metadata IP blocklist + if host in _SSRF_METADATA_IPS: + raise ValueError(f"URL target '{host}' is a cloud metadata endpoint") + + # 3. Scheme validation + scheme = (parsed.scheme or "").lower() + if scheme not in ("http", "https"): + raise ValueError(f"URL scheme '{scheme}' is not supported") + + # 4. IP-based check for literal addresses + try: + ip = ipaddress.ip_address(host) + except ValueError: + pass # not an IP literal — continue to DNS check below + else: + if ip.is_private or ip.is_loopback or ip.is_link_local: + raise ValueError(f"URL target '{host}' is a private or reserved IP") + if host in _SSRF_METADATA_IPS: + raise ValueError(f"URL target '{host}' is a cloud metadata endpoint") + return # IP literal passed validation + + # 5. DNS resolution check for hostnames + try: + addr_info = socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM) + for _, _, _, _, sockaddr in addr_info: + ip_str = sockaddr[0] + ip = ipaddress.ip_address(ip_str) + if ip.is_private or ip.is_loopback or ip.is_link_local: + raise ValueError( + f"URL target '{host}' resolves to private IP '{ip_str}'" + ) + if ip_str in _SSRF_METADATA_IPS: + raise ValueError( + f"URL target '{host}' resolves to metadata IP '{ip_str}'" + ) + except OSError: + # DNS failure — let the caller handle transient errors + pass + class WebChannel(Channel): name = "web" @@ -22,9 +92,19 @@ def check(self, config=None): return "ok", "通过 Jina Reader 读取任意网页（curl https://r.jina.ai/URL）" def read(self, url: str) -> str: - """通过 Jina Reader 读取网页，返回 Markdown 全文。""" + """通过 Jina Reader 读取网页，返回 Markdown 全文。 + + Validates the target URL to block SSRF to private/internal hosts. + """ if not url.startswith(("http://", "https://")): url = "https://" + url + + # SSRF protection: reject private/internal targets before proxying. + try: + _validate_url(url) + except ValueError as e: + raise RuntimeError(f"URL validation failed: {e}") from e + jina_url = f"https://r.jina.ai/{url}" req = urllib.request.Request( jina_url, diff --git a/agent_reach/config.py b/agent_reach/config.py index 4386bb47..e772743d 100644 --- a/agent_reach/config.py +++ b/agent_reach/config.py @@ -3,15 +3,55 @@ Stores settings in ~/.agent-reach/config.yaml. Auto-creates directory on first use. +Sensitive values (tokens, keys, cookies) are encrypted at rest. """ +import base64 +import hashlib +import hmac import os +import stat from pathlib import Path from typing import Any, Optional import yaml +#: Config keys whose values are encrypted at rest. +#: Matched by lowercase substring — add new sensitive keys here. +_SENSITIVE_KEY_PATTERNS = ( + "key", "token", "password", "secret", + "cookie", "auth", "proxy", +) + +#: Environment variable names that Config.get() may fall back to. +#: All other uppercase env var names are ignored for safety. +_ENV_ALLOWLIST = frozenset({ + # API keys + "GROQ_API_KEY", + "OPENAI_API_KEY", + "EXA_API_KEY", + "GITHUB_TOKEN", + # Auth tokens / cookies + "TWITTER_AUTH_TOKEN", + "TWITTER_CT0", + "BILIBILI_SESSDATA", + "BILIBILI_CSRF", + "XHS_COOKIE", + "XUEQIU_COOKIE", + "YOUTUBE_COOKIES_FROM", + # Proxy / environment + "PROXY", + "AGENT_REACH_LANG", +}) + +#: Encryption marker prefix — values starting with this use at-rest encryption. +_ENC_PREFIX = "$enc$" +_ENC_SALT = b"agent-reach-enc-v1" +_ENC_KEYGEN_ITERATIONS = 200_000 +_ENC_KEYFILE = ".config_key" + + class Config: """Manages Agent Reach configuration.""" @@ -38,47 +78,146 @@ def _ensure_dir(self): """Create config directory if it doesn't exist.""" self.config_dir.mkdir(parents=True, exist_ok=True) + # ── Encryption helpers ────────────────────────────────────── + + @staticmethod + def _is_sensitive(key: str) -> bool: + """True if *key* holds a credential that should be encrypted at rest.""" + kl = key.lower() + return any(pattern in kl for pattern in _SENSITIVE_KEY_PATTERNS) + + @staticmethod + def _key_path() -> Path: + return Config.CONFIG_DIR / _ENC_KEYFILE + + @staticmethod + def _load_or_create_key() -> bytes: + """Load the machine-local encryption key, creating it on first use.""" + kp = Config._key_path() + if kp.exists(): + return kp.read_bytes() + # Create with 0o600 so the key is never world-readable. + key = os.urandom(32) + try: + fd = os.open(str(kp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, + stat.S_IRUSR | stat.S_IWUSR) + with os.fdopen(fd, "wb") as f: + f.write(key) + except OSError: + with open(kp, "wb") as f: + f.write(key) + try: + os.chmod(kp, 0o600) + except OSError: + pass + return key + + def _encrypt_value(self, plaintext: str) -> str: + """Encrypt *plaintext* for at-rest storage. + + Uses PBKDF2-derived key material + HMAC-SHA256 integrity. + This is NOT production-grade encryption — install ``cryptography`` + (Fernet) for stronger protection. The goal here is to prevent + casual credential disclosure from config file reads. + """ + key = self._load_or_create_key() + salt = os.urandom(16) + dk = hashlib.pbkdf2_hmac("sha256", key, salt + _ENC_SALT, + _ENC_KEYGEN_ITERATIONS, dklen=64) + enc_key, mac_key = dk[:32], dk[32:] + # Simple XOR keystream (enc_key as seed — not AES, see docstring). + plain_bytes = plaintext.encode("utf-8") + keystream = hashlib.sha256(enc_key + salt).digest() + while len(keystream) < len(plain_bytes): + keystream += hashlib.sha256(enc_key + keystream[-32:]).digest() + cipher = bytes(a ^ b for a, b in zip(plain_bytes, keystream[:len(plain_bytes)])) + tag = hmac.new(mac_key, salt + cipher, "sha256").digest() + return _ENC_PREFIX + base64.b64encode(salt + tag + cipher).decode("ascii") + + def _decrypt_value(self, ciphertext: str) -> str: + """Decrypt a value previously encrypted with ``_encrypt_value``.""" + if not ciphertext.startswith(_ENC_PREFIX): + return ciphertext # not encrypted (backward compat) + raw = base64.b64decode(ciphertext[len(_ENC_PREFIX):]) + salt, tag, cipher = raw[:16], raw[16:48], raw[48:] + key = self._load_or_create_key() + dk = hashlib.pbkdf2_hmac("sha256", key, salt + _ENC_SALT, + _ENC_KEYGEN_ITERATIONS, dklen=64) + enc_key, mac_key = dk[:32], dk[32:] + expected = hmac.new(mac_key, salt + cipher, "sha256").digest() + if not hmac.compare_digest(tag, expected): + raise ValueError("config integrity check failed — key file may have changed") + keystream = hashlib.sha256(enc_key + salt).digest() + while len(keystream) < len(cipher): + keystream += hashlib.sha256(enc_key + keystream[-32:]).digest() + return bytes(a ^ b for a, b in zip(cipher, keystream[:len(cipher)])).decode("utf-8") + + # ── Load / Save ───────────────────────────────────────────── + def load(self): - """Load config from YAML file.""" + """Load config from YAML file, decrypting sensitive fields.""" if self.config_path.exists(): with open(self.config_path, "r", encoding="utf-8") as f: - self.data = yaml.safe_load(f) or {} + raw = yaml.safe_load(f) or {} + # Decrypt any encrypted values transparently. + self.data = { + k: (self._decrypt_value(v) if isinstance(v, str) else v) + for k, v in raw.items() + } else: self.data = {} def save(self): - """Save config to YAML file.""" + """Save config to YAML file, encrypting sensitive fields.""" self._ensure_dir() - # Create file with restricted permissions from the start to avoid - # a race window where credentials are briefly world-readable. + # Encrypt sensitive values before writing. + to_write = {} + for k, v in self.data.items(): + if isinstance(v, str) and self._is_sensitive(k): + to_write[k] = self._encrypt_value(v) + else: + to_write[k] = v + # Create file with restricted permissions. try: - import stat fd = os.open( str(self.config_path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, stat.S_IRUSR | stat.S_IWUSR, # 0o600 ) with os.fdopen(fd, "w", encoding="utf-8") as f: - yaml.dump(self.data, f, default_flow_style=False, allow_unicode=True) + yaml.dump(to_write, f, default_flow_style=False, allow_unicode=True) except OSError: # Fallback for Windows or other edge cases where os.open flags # are not fully supported. with open(self.config_path, "w", encoding="utf-8") as f: - yaml.dump(self.data, f, default_flow_style=False, allow_unicode=True) + yaml.dump(to_write, f, default_flow_style=False, allow_unicode=True) + + # ── Accessors ─────────────────────────────────────────────── def get(self, key: str, default: Any = None) -> Any: - """Get a config value. Also checks environment variables (uppercase).""" - # Config file first + """Get a config value. + + Checks (in order): + 1. Config file (decrypted transparently on load). + 2. Environment variable — *only* if *key* appears in ``_ENV_ALLOWLIST``. + + This prevents accidental leakage of unrelated environment variables. + """ if key in self.data: return self.data[key] - # Then env var (uppercase) - env_val = os.environ.get(key.upper()) - if env_val: - return env_val + # Only fall back to env var for allowlisted names. + env_key = key.upper() + if env_key in _ENV_ALLOWLIST: + env_val = os.environ.get(env_key) + if env_val: + return env_val return default def set(self, key: str, value: Any): - """Set a config value and save.""" + """Set a config value and save. + + Sensitive values (tokens, keys, cookies) are encrypted at rest. + """ self.data[key] = value self.save() diff --git a/agent_reach/cookie_extract.py b/agent_reach/cookie_extract.py index 545f9621..f81f5f5b 100644 --- a/agent_reach/cookie_extract.py +++ b/agent_reach/cookie_extract.py @@ -44,7 +44,12 @@ def extract_all(browser: str = "chrome") -> Dict[str, dict]: """ Extract cookies for all supported platforms from the specified browser. - + + NOTE: This reads cookies for the configured browser profile. Only cookies + matching the target platform domains (Twitter/X, XiaoHongShu, Bilibili, + Xueqiu) are retained — other sites' cookies are not stored. However, the + underlying cookie library may still load all cookies into memory briefly. + Returns: { "twitter": {"auth_token": "xxx", "ct0": "yyy"}, diff --git a/agent_reach/transcribe.py b/agent_reach/transcribe.py index 2ffd6124..93d6fce5 100644 --- a/agent_reach/transcribe.py +++ b/agent_reach/transcribe.py @@ -24,6 +24,12 @@ from agent_reach.config import Config # Whisper API limit is 25MB; leave headroom for multipart overhead. +# Common audio file extensions accepted for local file transcription. +_AUDIO_EXTENSIONS = frozenset({ + ".wav", ".mp3", ".m4a", ".flac", ".ogg", ".opus", + ".aac", ".wma", ".aiff", ".webm", +}) + SIZE_LIMIT_BYTES = 24 * 1024 * 1024 CHUNK_SECONDS = 600 # 10 min — small enough that boundary cuts rarely lose meaning @@ -88,6 +94,7 @@ def download_audio(url: str, out_dir: Path) -> Path: "0", "-o", str(template), + "--", # guard against url injection as yt-dlp option url, ], timeout=1800, # long podcasts over slow networks — generous but bounded @@ -229,6 +236,13 @@ def transcribe( src_path = Path(source) if src_path.is_file(): + # Local file path: validate it looks like audio before processing. + if src_path.suffix.lower() not in _AUDIO_EXTENSIONS: + raise TranscribeError( + f"unsupported file type '{src_path.suffix}': " + f"only audio files ({', '.join(sorted(_AUDIO_EXTENSIONS))}) " + f"can be transcribed. For non-audio files, use a URL instead." + ) audio = src_path else: audio = download_audio(source, work_dir) diff --git a/tests/test_config.py b/tests/test_config.py index a7bb2a42..bf90276f 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -33,13 +33,13 @@ def test_get_default(self, tmp_config): assert tmp_config.get("nonexistent", "default") == "default" def test_get_from_env(self, tmp_config, monkeypatch): - monkeypatch.setenv("TEST_ENV_KEY", "env_value") - assert tmp_config.get("test_env_key") == "env_value" + monkeypatch.setenv("EXA_API_KEY", "env_value") + assert tmp_config.get("exa_api_key") == "env_value" def test_config_file_priority_over_env(self, tmp_config, monkeypatch): - monkeypatch.setenv("MY_KEY", "from_env") - tmp_config.set("my_key", "from_config") - assert tmp_config.get("my_key") == "from_config" + monkeypatch.setenv("EXA_API_KEY", "from_env") + tmp_config.set("exa_api_key", "from_config") + assert tmp_config.get("exa_api_key") == "from_config" def test_save_and_load(self, tmp_config): tmp_config.set("key1", "value1")