fix(security): block all bare email tool names for non-admins; harden fence-tag regex

Review follow-up on #3681 (thanks @vgalin):

1. Routing bare email names made 10 of the 14 email tools executable by
   non-admin owners — is_public_blocked_tool() runs on the bare name
   before dispatch, and NON_ADMIN_BLOCKED_TOOLS only listed 4. Define the
   full email tool set once (BUILTIN_EMAIL_TOOLS in tool_security.py) and
   derive the blocklist, the fence tags (TOOL_TAGS), the bare-name
   dispatch, and the native-call mapping from it so they can't drift.
   This also fixes 4 tools (search_emails, draft_email, draft_email_reply,
   ai_draft_email_reply) that were missing from the old tool_schemas copy
   and therefore unreachable even for native function-calling models.

2. The relaxed fence regex from the previous commit could prefix-match
   longer fence tags: ```python3 parsed as tool "python" with content
   "3\nprint(...)" and executed as code. Add a (?![\w-]) boundary after
   the tag.

Tests: test_public_agent_policy_blocks_sensitive_tools now covers all 14
bare email names + the mcp__email__ form; new tests/test_fenced_inline_args.py
pins inline-args parsing, the python3/hyphenated-tag non-matches, and
strip/parse display mirroring.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
botinate
2026-06-11 01:27:54 -07:00
parent f24dbadab3
commit 544a664766
7 changed files with 128 additions and 22 deletions
+6 -4
View File
@@ -14,6 +14,7 @@ Sub-modules:
import logging
from collections import namedtuple
from src.tool_security import BUILTIN_EMAIL_TOOLS
from src.tool_utils import _truncate, get_mcp_manager, set_mcp_manager
logger = logging.getLogger(__name__)
@@ -64,9 +65,10 @@ TOOL_TAGS = {"bash", "python", "web_search", "web_fetch", "read_file", "write_fi
"manage_endpoints", "manage_mcp", "manage_webhooks",
"manage_tokens", "manage_documents", "manage_settings",
"manage_notes", "manage_calendar",
"resolve_contact", "manage_contact", "list_email_accounts", "send_email", "list_emails",
"read_email", "reply_to_email", "bulk_email", "archive_email",
"delete_email", "mark_email_read",
"resolve_contact", "manage_contact",
# Email tool names come from BUILTIN_EMAIL_TOOLS (unioned below)
# so the fence regex, dispatch, and non-admin blocklist all cover
# the same set.
# Cookbook tools (LLM serving + downloads). Without these
# entries, native function calls to e.g. list_served_models
# are rejected as "Unknown function call" before reaching
@@ -83,7 +85,7 @@ TOOL_TAGS = {"bash", "python", "web_search", "web_fetch", "read_file", "write_fi
# Generic loopback to any UI-button endpoint (cookbook,
# gallery, email folders, etc.) — agent uses this when
# there's no named tool wrapper for the action.
"app_api"}
"app_api"} | BUILTIN_EMAIL_TOOLS
ToolBlock = namedtuple("ToolBlock", ["tool_type", "content"])
+5 -6
View File
@@ -20,7 +20,7 @@ from typing import Any, Awaitable, Callable, Dict, Optional, Tuple
from src.tool_security import is_public_blocked_tool, owner_is_admin_or_single_user
from src.tool_security import BUILTIN_EMAIL_TOOLS, is_public_blocked_tool, owner_is_admin_or_single_user
from src.tool_policy import ToolPolicy
from src.constants import MAX_OUTPUT_CHARS, MAX_READ_CHARS, MAX_DIFF_LINES, DATA_DIR
from src.tool_utils import _truncate, get_mcp_manager
@@ -767,11 +767,10 @@ async def execute_tool_block(
elif tool == "vault_unlock":
desc = "vault_unlock"
result = await do_vault_unlock(content, owner=owner)
elif tool in {"list_email_accounts", "send_email", "list_emails", "read_email",
"reply_to_email", "bulk_email", "archive_email", "delete_email",
"mark_email_read", "search_emails", "draft_email", "draft_email_reply",
"ai_draft_email_reply", "download_attachment"}:
# Bare email tool name from fenced-block models (e.g. Ollama) — route to MCP email server
elif tool in BUILTIN_EMAIL_TOOLS:
# Bare email tool name from fenced-block models (e.g. Ollama) — route to MCP email server.
# Non-admin owners never reach here: BUILTIN_EMAIL_TOOLS ⊆ NON_ADMIN_BLOCKED_TOOLS,
# so is_public_blocked_tool() above already rejected them.
mcp = get_mcp_manager()
qualified = f"mcp__email__{tool}"
if mcp:
+6 -2
View File
@@ -19,9 +19,13 @@ logger = logging.getLogger(__name__)
# Regex patterns
# ---------------------------------------------------------------------------
# Pattern 1: ```bash ... ``` fenced code blocks
# Pattern 1: ```bash ... ``` fenced code blocks. The tag may be followed by
# inline args on the same line (```list_email_accounts {}) or a newline.
# (?![\w-]) keeps the alternation from prefix-matching longer fence tags:
# without it, ```python3 would match as tool "python" with content "3\n..."
# and execute as code.
_TOOL_BLOCK_RE = re.compile(
r"```(" + "|".join(TOOL_TAGS) + r")[ \t]*\n?([\s\S]*?)```",
r"```(" + "|".join(TOOL_TAGS) + r")(?![\w-])[ \t]*\n?([\s\S]*?)```",
re.IGNORECASE,
)
+2 -3
View File
@@ -14,6 +14,7 @@ from typing import Optional
from src.agent_tools import ToolBlock, TOOL_TAGS
from src.tool_parsing import _TOOL_NAME_MAP
from src.tool_security import BUILTIN_EMAIL_TOOLS
logger = logging.getLogger(__name__)
@@ -1211,9 +1212,7 @@ def function_call_to_tool_block(name: str, arguments: str) -> Optional[ToolBlock
content = json.dumps(args) if args else "{}"
return ToolBlock(tool_type, content)
# Email tools are implemented as MCP — route them to email
_BUILTIN_EMAIL_TOOLS = {"list_email_accounts", "send_email", "list_emails", "read_email", "reply_to_email",
"archive_email", "delete_email", "mark_email_read", "bulk_email", "download_attachment"}
if name in _BUILTIN_EMAIL_TOOLS:
if name in BUILTIN_EMAIL_TOOLS:
return ToolBlock(f"mcp__email__{name}", json.dumps(args) if args else "{}")
if tool_type not in TOOL_TAGS:
logger.warning(f"Unknown function call: {name}")
+28 -6
View File
@@ -8,10 +8,36 @@ from typing import Optional, Set
logger = logging.getLogger(__name__)
# Every tool exposed by the built-in email MCP server
# (mcp_servers/email_server.py). Single source of truth: the fence tags
# (TOOL_TAGS), bare-name dispatch (tool_execution), native-call mapping
# (tool_schemas), and the non-admin blocklist below all derive from this set,
# so a tool added to the email server can't become reachable under its bare
# name without also being blocked for non-admins.
BUILTIN_EMAIL_TOOLS = frozenset({
"list_email_accounts",
"list_emails",
"read_email",
"search_emails",
"send_email",
"reply_to_email",
"draft_email",
"draft_email_reply",
"ai_draft_email_reply",
"archive_email",
"delete_email",
"mark_email_read",
"bulk_email",
"download_attachment",
})
# Tools regular/public users must not execute directly. These either expose
# server/runtime access, sensitive user data, external messaging, persistent
# state changes, or generic loopback/integration surfaces.
NON_ADMIN_BLOCKED_TOOLS = {
# state changes, or generic loopback/integration surfaces. All email tools are
# included (SECURITY.md: email/MCP capabilities are privileged admin
# functionality).
NON_ADMIN_BLOCKED_TOOLS = BUILTIN_EMAIL_TOOLS | {
"bash",
"python",
"read_file",
@@ -32,10 +58,6 @@ NON_ADMIN_BLOCKED_TOOLS = {
"manage_settings",
"api_call",
"app_api",
"send_email",
"reply_to_email",
"list_emails",
"read_email",
"resolve_contact",
"manage_contact",
"manage_calendar",
+71
View File
@@ -0,0 +1,71 @@
"""PR #3681 — fenced tool calls with inline args, and the fence-tag boundary.
Local fenced-block models (Ollama etc.) emit calls like ```list_email_accounts {}
with the args on the same line as the tag; the parser must execute those. The
relaxed tag pattern must NOT prefix-match longer fence tags: ```python3 is a
language hint, not a "python" tool call with content "3\n...".
"""
import sys
from unittest.mock import MagicMock
for mod in ['src.agent_tools', 'src.tool_parsing', 'src.tool_schemas', 'src.tool_execution']:
sys.modules.pop(mod, None)
for mod in [
'sqlalchemy', 'sqlalchemy.orm', 'sqlalchemy.ext', 'sqlalchemy.ext.declarative',
'sqlalchemy.ext.hybrid', 'sqlalchemy.sql', 'sqlalchemy.sql.expression',
'src.database', 'core.models', 'core.database', 'core.auth'
]:
if mod not in sys.modules:
sys.modules[mod] = MagicMock()
import src.agent_tools # noqa: E402, F401
from src.tool_parsing import parse_tool_blocks, strip_tool_blocks # noqa: E402
def test_inline_args_on_tag_line_parse():
# The original bug: ```list_email_accounts {} (args on the tag line)
# never matched because the regex required a newline right after the tag.
blocks = parse_tool_blocks('```list_email_accounts {}\n```')
assert [(b.tool_type, b.content) for b in blocks] == [("list_email_accounts", "{}")]
def test_inline_json_args_parse_for_email_tools():
blocks = parse_tool_blocks('```list_emails {"max_results": 5}\n```')
assert [(b.tool_type, b.content) for b in blocks] == [("list_emails", '{"max_results": 5}')]
def test_next_line_content_still_parses():
# No regression for the classic shape: tag, newline, content.
blocks = parse_tool_blocks('```manage_memory\nadd\nsome text\n```')
assert [(b.tool_type, b.content) for b in blocks] == [("manage_memory", "add\nsome text")]
def test_plain_bash_fence_still_parses():
blocks = parse_tool_blocks('```bash\necho hello\n```')
assert [(b.tool_type, b.content) for b in blocks] == [("bash", "echo hello")]
def test_python3_language_hint_is_not_a_python_tool_call():
# ```python3 must not prefix-match the "python" fence tag — without the
# (?![\w-]) boundary it parsed as tool "python" with content "3\nprint(...)"
# and executed as code.
blocks = parse_tool_blocks('```python3\nprint("hi")\n```')
assert blocks == [], blocks
def test_hyphenated_tag_is_not_a_tool_call():
blocks = parse_tool_blocks('```bash-session\n$ ls\n```')
assert blocks == [], blocks
def test_inline_args_fence_is_stripped_from_display():
# strip must mirror parse: an executed inline-args fence must not leak
# into the displayed text.
text = 'Checking now.\n```list_email_accounts {}\n```\nDone.'
assert strip_tool_blocks(text) == 'Checking now.\n\nDone.'
def test_python3_fence_is_left_intact_in_display():
# ...and a fence that did NOT parse as a tool call must stay visible.
text = 'Example:\n```python3\nprint("hi")\n```'
assert strip_tool_blocks(text) == text
+10 -1
View File
@@ -616,7 +616,16 @@ async def test_public_agent_policy_blocks_sensitive_tools(monkeypatch):
monkeypatch.setattr(auth_mod, "AuthManager", lambda: FakeAuth())
for tool_name in ("send_email", "read_file", "mcp__email__send_email"):
# Every bare email tool name is spelled out (not imported from
# BUILTIN_EMAIL_TOOLS) so accidentally dropping one from that set fails
# here instead of silently shrinking the blocklist.
bare_email_tools = (
"list_email_accounts", "list_emails", "read_email", "search_emails",
"send_email", "reply_to_email", "draft_email", "draft_email_reply",
"ai_draft_email_reply", "archive_email", "delete_email",
"mark_email_read", "bulk_email", "download_attachment",
)
for tool_name in bare_email_tools + ("read_file", "mcp__email__send_email"):
desc, result = await execute_tool_block(
SimpleNamespace(tool_type=tool_name, content="{}"),
owner="regular-user",