Security hardening: credential storage, input validation, and access controls

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,

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()
 

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"},

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)

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")