mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-07-17 11:07:11 +08:00
Merge branch 'master' of github.com:zhayujie/chatgpt-on-wechat
This commit is contained in:
@@ -4,6 +4,7 @@ Supports text files, images (jpg, png, gif, webp), and PDF files
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import os
|
import os
|
||||||
|
import re
|
||||||
from typing import Dict, Any
|
from typing import Dict, Any
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
@@ -12,6 +13,12 @@ from agent.tools.utils.truncate import truncate_head, format_size, DEFAULT_MAX_L
|
|||||||
from common.utils import expand_path
|
from common.utils import expand_path
|
||||||
|
|
||||||
|
|
||||||
|
# Paths whose CONTENT mirrors the process environment (and thus any secrets
|
||||||
|
# loaded from ~/.cow/.env). Reading them bypasses the env_config boundary.
|
||||||
|
# Matches /proc/self/environ, /proc/thread-self/environ and /proc/<pid>/environ.
|
||||||
|
_PROC_ENVIRON_RE = re.compile(r"^/proc/(\d+|self|thread-self)/environ$")
|
||||||
|
|
||||||
|
|
||||||
class Read(BaseTool):
|
class Read(BaseTool):
|
||||||
"""Tool for reading file contents"""
|
"""Tool for reading file contents"""
|
||||||
|
|
||||||
@@ -79,9 +86,9 @@ class Read(BaseTool):
|
|||||||
# Resolve path
|
# Resolve path
|
||||||
absolute_path = self._resolve_path(path)
|
absolute_path = self._resolve_path(path)
|
||||||
|
|
||||||
# Security check: Prevent reading sensitive config files
|
# Security check: block credential files and their aliases.
|
||||||
env_config_path = expand_path("~/.cow/.env")
|
# See issue #2913 (/proc/self/environ bypass) and #2863 (scope).
|
||||||
if os.path.abspath(absolute_path) == os.path.abspath(env_config_path):
|
if self._is_credential_path(absolute_path):
|
||||||
return ToolResult.fail(
|
return ToolResult.fail(
|
||||||
"Error: Access denied. API keys and credentials must be accessed through the env_config tool only."
|
"Error: Access denied. API keys and credentials must be accessed through the env_config tool only."
|
||||||
)
|
)
|
||||||
@@ -141,6 +148,38 @@ class Read(BaseTool):
|
|||||||
return path
|
return path
|
||||||
return os.path.abspath(os.path.join(self.cwd, path))
|
return os.path.abspath(os.path.join(self.cwd, path))
|
||||||
|
|
||||||
|
def _is_credential_path(self, absolute_path: str) -> bool:
|
||||||
|
"""Return True if *absolute_path* points at protected credential data.
|
||||||
|
|
||||||
|
Beyond the literal ~/.cow/.env file, this also blocks two real bypass
|
||||||
|
surfaces reported in issue #2913:
|
||||||
|
1. /proc/<pid|self|thread-self>/environ — a second view of the
|
||||||
|
process environment that leaks secrets loaded from ~/.cow/.env.
|
||||||
|
2. Symlinks resolving to ~/.cow/.env; the previous exact abspath
|
||||||
|
match kept the link target and could be bypassed.
|
||||||
|
|
||||||
|
Scope is kept deliberately narrow (only the credential file and its
|
||||||
|
environ aliases) so this does NOT re-broaden the block that #2863
|
||||||
|
intentionally narrowed to ~/.cow/.env.
|
||||||
|
"""
|
||||||
|
# Compare on both the normalized path and the symlink-resolved path,
|
||||||
|
# in POSIX form so the /proc regex matches regardless of os.sep.
|
||||||
|
candidates = set()
|
||||||
|
try:
|
||||||
|
candidates.add(os.path.normpath(absolute_path).replace(os.sep, "/"))
|
||||||
|
candidates.add(os.path.realpath(absolute_path).replace(os.sep, "/"))
|
||||||
|
except OSError:
|
||||||
|
candidates.add(absolute_path.replace(os.sep, "/"))
|
||||||
|
|
||||||
|
# 1. /proc environ aliases (checked on raw and symlink-resolved forms).
|
||||||
|
for candidate in candidates:
|
||||||
|
if _PROC_ENVIRON_RE.match(candidate):
|
||||||
|
return True
|
||||||
|
|
||||||
|
# 2. The credential file itself, following symlinks on both sides.
|
||||||
|
env_real = os.path.realpath(expand_path("~/.cow/.env")).replace(os.sep, "/")
|
||||||
|
return env_real in candidates
|
||||||
|
|
||||||
def _return_file_metadata(self, absolute_path: str, file_type: str, file_size: int) -> ToolResult:
|
def _return_file_metadata(self, absolute_path: str, file_type: str, file_size: int) -> ToolResult:
|
||||||
"""
|
"""
|
||||||
Return file metadata for non-readable files (video, audio, binary, etc.)
|
Return file metadata for non-readable files (video, audio, binary, etc.)
|
||||||
|
|||||||
103
tests/test_security_read_env_bypass.py
Normal file
103
tests/test_security_read_env_bypass.py
Normal file
@@ -0,0 +1,103 @@
|
|||||||
|
# encoding:utf-8
|
||||||
|
"""
|
||||||
|
Unit tests for Read tool credential-bypass protection (issue #2913).
|
||||||
|
|
||||||
|
Verifies that the read tool blocks not just the literal ~/.cow/.env file but
|
||||||
|
also its process-environment aliases (/proc/<pid>/environ) and symlinks that
|
||||||
|
resolve to it, WITHOUT re-broadening the scope narrowed by #2863.
|
||||||
|
"""
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||||
|
|
||||||
|
from agent.tools.read.read import Read
|
||||||
|
from common.utils import expand_path
|
||||||
|
|
||||||
|
_DENIED = "Access denied"
|
||||||
|
|
||||||
|
|
||||||
|
class TestReadCredentialBypass(unittest.TestCase):
|
||||||
|
"""_is_credential_path must block credential files and their aliases."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.read = Read()
|
||||||
|
|
||||||
|
# ---- happy path: ordinary files are NOT blocked -----------------------
|
||||||
|
|
||||||
|
def test_normal_file_not_blocked(self):
|
||||||
|
"""A regular temp file must not be treated as a credential path."""
|
||||||
|
with tempfile.NamedTemporaryFile(suffix=".txt", delete=False) as tf:
|
||||||
|
tf.write(b"hello world")
|
||||||
|
tmp_path = tf.name
|
||||||
|
try:
|
||||||
|
self.assertFalse(self.read._is_credential_path(tmp_path))
|
||||||
|
result = self.read.execute({"path": tmp_path})
|
||||||
|
self.assertEqual(result.status, "success")
|
||||||
|
finally:
|
||||||
|
os.unlink(tmp_path)
|
||||||
|
|
||||||
|
def test_non_environ_proc_file_not_blocked(self):
|
||||||
|
"""Non-environ /proc files must not be over-blocked (no #2863 regression)."""
|
||||||
|
self.assertFalse(self.read._is_credential_path("/proc/self/status"))
|
||||||
|
self.assertFalse(self.read._is_credential_path("/proc/1/cmdline"))
|
||||||
|
|
||||||
|
# ---- control: the literal credential file stays blocked --------------
|
||||||
|
|
||||||
|
def test_direct_env_file_blocked(self):
|
||||||
|
"""Direct ~/.cow/.env access must remain blocked (regression control)."""
|
||||||
|
env_path = expand_path("~/.cow/.env")
|
||||||
|
self.assertTrue(self.read._is_credential_path(env_path))
|
||||||
|
result = self.read.execute({"path": env_path})
|
||||||
|
self.assertEqual(result.status, "error")
|
||||||
|
self.assertIn(_DENIED, str(result.result))
|
||||||
|
|
||||||
|
# ---- the vulnerability: environ aliases are blocked ------------------
|
||||||
|
|
||||||
|
def test_proc_self_environ_blocked(self):
|
||||||
|
"""/proc/self/environ must be blocked (the #2913 bypass)."""
|
||||||
|
self.assertTrue(self.read._is_credential_path("/proc/self/environ"))
|
||||||
|
result = self.read.execute({"path": "/proc/self/environ"})
|
||||||
|
self.assertEqual(result.status, "error")
|
||||||
|
self.assertIn(_DENIED, str(result.result))
|
||||||
|
|
||||||
|
def test_proc_pid_environ_blocked(self):
|
||||||
|
"""/proc/<pid>/environ must be blocked for any pid."""
|
||||||
|
self.assertTrue(self.read._is_credential_path("/proc/1/environ"))
|
||||||
|
self.assertTrue(self.read._is_credential_path(f"/proc/{os.getpid()}/environ"))
|
||||||
|
|
||||||
|
def test_proc_thread_self_environ_blocked(self):
|
||||||
|
"""/proc/thread-self/environ must be blocked."""
|
||||||
|
self.assertTrue(self.read._is_credential_path("/proc/thread-self/environ"))
|
||||||
|
|
||||||
|
# ---- symlink escape --------------------------------------------------
|
||||||
|
|
||||||
|
@unittest.skipUnless(hasattr(os, "symlink"), "symlink not supported")
|
||||||
|
def test_symlink_to_env_file_blocked(self):
|
||||||
|
"""A symlink resolving to ~/.cow/.env must be blocked."""
|
||||||
|
env_path = expand_path("~/.cow/.env")
|
||||||
|
os.makedirs(os.path.dirname(env_path), exist_ok=True)
|
||||||
|
created_env = False
|
||||||
|
if not os.path.exists(env_path):
|
||||||
|
with open(env_path, "w", encoding="utf-8") as f:
|
||||||
|
f.write("SECRET=canary\n")
|
||||||
|
created_env = True
|
||||||
|
link_dir = tempfile.mkdtemp()
|
||||||
|
link_path = os.path.join(link_dir, "innocent.txt")
|
||||||
|
try:
|
||||||
|
os.symlink(env_path, link_path)
|
||||||
|
except (OSError, NotImplementedError):
|
||||||
|
self.skipTest("cannot create symlink in this environment")
|
||||||
|
try:
|
||||||
|
self.assertTrue(self.read._is_credential_path(link_path))
|
||||||
|
finally:
|
||||||
|
os.unlink(link_path)
|
||||||
|
os.rmdir(link_dir)
|
||||||
|
if created_env:
|
||||||
|
os.unlink(env_path)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user