Keep fixed values, tool type and connected servers out of editors' tool saves

/api/update_tool now checks submitted actions against the stored ones the
same way /api/update_tool_actions does: nothing can be added, and only the
tool's owner can change a fixed value. The tool type (name) can no longer
be changed after creation by anyone.

For API tools, changing who fills an existing header, query or body
parameter, its value, or clearing a stored value is now owner-only on both
/api/update_tool and /api/update_tool_config. Before, an editor could hand
a stored secret query value to the model and read it back in the chat.
The chat now masks every value that came from the stored action rather
than from the model.

A connection-backed MCP tool can no longer be moved to another server or
sign-in method through /api/update_tool or /api/update_tool_config (400,
pointing to /api/mcp_server/save), and a new key saved through
/api/update_tool_config goes to its connection, owner only.
This commit is contained in:
arc53-machine committed 2026-09-29 14:11:36 +01:00
1 parent a1789d2ab4
commit 4f5cd68771
4 files changed
+460 -10

No files matched your search

+13 -3
View File
@@ -115,11 +115,21 @@ def sent_arguments(
) -> dict:
"""What a call sends, flattened for the chat (the approval card, a finished call).
It follows :func:`resolve_arguments`, but a value the owner fixed shows
as :data:`FIXED_MASK`: it may be a secret (an API key in a query), and
It follows :func:`resolve_arguments`, but any value that comes from the
stored action rather than from the model shows as :data:`FIXED_MASK`:
a fixed value, and a default the model left alone, may be a secret (an
api_tool query value is decrypted into the action before the call), and
the chat is shown to whoever runs the agent, over the API or a widget
too. A value the connection sets (Telegram's default chat) is the
account's own setting and shows as it is. Headers are left out.
Args:
action: The stored action, with any secrets merged back.
llm_arguments: The arguments the model sent.
connection_pins: Parameter name to a value the connection fixes.
Returns:
Parameter name to the value shown for it.
"""
connection_pins = connection_pins or {}
shown: dict = {}
@@ -134,7 +144,7 @@ def sent_arguments(
elif name in llm_arguments:
shown[name] = llm_arguments[name]
elif has_value(details):
shown[name] = details["value"]
shown[name] = FIXED_MASK
return shown
+176 -7
View File
@@ -19,7 +19,7 @@ from docsgpt.agents.default_tools import (
WORKFLOW_ONLY_BUILTINS,
)
from docsgpt.agents.tool_executor import API_TOOL_SECRET_SECTIONS, API_TOOL_SECRETS_KEY
from docsgpt.agents.tool_pins import merge_submitted_actions, PinChangeRefused
from docsgpt.agents.tool_pins import iter_parameters, llm_fills, merge_submitted_actions, PinChangeRefused
from docsgpt.agents.tools.spec_parser import parse_spec
from docsgpt.agents.tools.tool_manager import ToolManager
from docsgpt.api import api
@@ -34,7 +34,7 @@ from docsgpt.api.user.resource_access import (
settings_many,
)
from docsgpt.api.user.team_sharing import visible_with_access
from docsgpt.connectors.catalog import definition_for_tool
from docsgpt.connectors.catalog import base_url, definition_for_tool
from docsgpt.connectors.service import account_tool_names
from docsgpt.core.settings import settings
from docsgpt.core.url_validation import SSRFError, validate_url
@@ -192,7 +192,14 @@ def _merge_secrets_on_update(new_config, existing_config, config_requirements, u
_CREDENTIALS_FOR_NEW_SERVER = "Enter credentials for the new server"
_FORBIDDEN_MESSAGE = "Your access to this item doesn't allow that"
_MCP_CREDENTIAL_AUTH_TYPES = {"api_key", "bearer", "basic"}
_META_KEYS = ("name", "displayName", "customName", "description", "actions")
# ``name`` (the tool type) and ``actions`` are handled on their own.
_META_KEYS = ("displayName", "customName", "description")
_TYPE_IS_FIXED = "A tool's type can't be changed"
_FIXED_VALUES_OWNER_ONLY = "Only the tool's owner can change fixed values"
_MOVE_THROUGH_MCP_SAVE = (
"This server signs in through a connection: change its address or sign-in by saving the "
"server again (/api/mcp_server/save)"
)
class CredentialsRequired(Exception):
@@ -362,6 +369,96 @@ def _api_tool_config_needs_credentials(new_config: dict, existing_config: dict)
return False
def _api_tool_param_state(section: str, spec: dict, *, stored: bool) -> tuple:
"""Who fills an api_tool parameter and which value it keeps.
A header / query value is only ever shown masked, so a stored one (sealed,
or legacy plaintext) reads as ``"stored"`` and any value a client sends
is a new one.
Args:
section: ``headers``, ``query_params`` or ``body``.
spec: The parameter's schema.
stored: Whether ``spec`` comes from the stored config.
Returns:
``(filled_by_llm, value marker)``; the marker is None without a value.
"""
value = spec.get("value")
if section in API_TOOL_SECRET_SECTIONS:
if _has_value(value):
marker: Any = "stored" if stored else ("new", value)
else:
marker = "stored" if spec.get("has_value") else None
else:
marker = value if _has_value(value) else None
return llm_fills(spec), marker
def _api_tool_fixed_values_changed(new_config: dict, existing_config: dict) -> bool:
"""Whether an api_tool save changes a fixed value of an existing action.
A fixed value may be a secret (an API key in a query) or where a call
goes, so changing who fills a parameter, or its value, or clearing a
stored one is the owner's. Flipping ``filled_by_llm`` on a stored secret
would hand it to the model and show it in the chat. A parameter that
carries no value can be added or removed freely; a new action brings a
URL and is judged by :func:`_api_tool_config_needs_credentials`.
Args:
new_config: The config the client sent.
existing_config: The stored config.
Returns:
True when any existing action's fixed values would differ.
"""
old_actions = (existing_config or {}).get("actions") or {}
new_actions = (new_config or {}).get("actions") or {}
if not isinstance(old_actions, dict) or not isinstance(new_actions, dict):
return bool(old_actions) or bool(new_actions)
for name, action in new_actions.items():
old = old_actions.get(name)
if not isinstance(old, dict) or not isinstance(action, dict):
continue
before = {(s, p): _api_tool_param_state(s, d, stored=True) for s, p, d in iter_parameters(old)}
after = {(s, p): _api_tool_param_state(s, d, stored=False) for s, p, d in iter_parameters(action)}
for key in before.keys() | after.keys():
if key in before and key in after:
if before[key] != after[key]:
return True
elif (before.get(key) or after.get(key))[1] is not None:
return True
return False
def _connection_server_moved(tool_doc: dict, new_config: Optional[dict]) -> bool:
"""Whether a config save re-points a connection-backed MCP tool.
The connection holds the key for one server and one way of signing in;
a tool moved on these routes would keep a connection that no longer
applies (it is refused at run time) and the new key would be stored
where nothing reads it. ``/api/mcp_server/save`` moves a server properly.
Args:
tool_doc: The stored ``user_tools`` row.
new_config: The incoming ``config``.
Returns:
True when the base URL or ``auth_type`` would change.
"""
if tool_doc.get("name") != "mcp_tool" or not tool_doc.get("connection_id"):
return False
existing = tool_doc.get("config") or {}
new_config = new_config if isinstance(new_config, dict) else {}
if "server_url" in new_config and base_url(str(new_config.get("server_url") or "").strip()) != base_url(
existing.get("server_url")
):
return True
return "auth_type" in new_config and (new_config.get("auth_type") or "none") != (
existing.get("auth_type") or "none"
)
def _mcp_origin_changed(new_config: dict, existing_config: dict) -> bool:
"""Whether a save moves an MCP server to another scheme, host or port."""
old_url = (existing_config or {}).get("server_url")
@@ -399,6 +496,24 @@ def check_oauth_mcp_owner_only(
raise AccessDenied(403, SHARED_OAUTH_OWNER_ONLY)
def check_api_tool_fixed_values(
ra: ResourceAccess, new_config: Optional[dict], existing_config: Optional[dict]
) -> None:
"""Keep an api_tool's fixed header, query and body values with its owner.
Args:
ra: The caller's access to the tool.
new_config: The incoming ``config``.
existing_config: The stored ``config``.
Raises:
AccessDenied: 403 when a non-owner changes a fixed value (see
:func:`_api_tool_fixed_values_changed`).
"""
if ra.access != "owner" and _api_tool_fixed_values_changed(new_config or {}, existing_config or {}):
raise AccessDenied(403, _FIXED_VALUES_OWNER_ONLY)
def _prepare_tool_config(tool_doc: dict, new_config: dict, config_requirements: dict) -> dict:
"""Validate-free merge of an incoming config with the stored one, as the owner.
@@ -882,6 +997,16 @@ class UpdateTool(Resource):
)
@api.doc(description="Update a tool by ID")
def post(self):
"""Update a tool's names, actions, config or chat switch.
The tool type (``name``) never changes. ``actions`` are checked
against the stored ones like ``/api/update_tool_actions``: nothing
is added and fixed values are the owner's. A connection-backed MCP
server is moved only through ``/api/mcp_server/save``.
Returns:
``{"success": true}``, or 400 / 403 / 404 with a message.
"""
decoded_token = request.decoded_token
if not decoded_token:
return make_response(jsonify({"success": False}), 401)
@@ -956,13 +1081,32 @@ class UpdateTool(Resource):
return make_response(
jsonify({"success": False, "message": "Tool not found"}), 404,
)
if update_data:
if "name" in data and data["name"] != tool_doc.get("name"):
# The type decides what the config and the connection's
# credentials are used for; it is set once, on create.
return make_response(jsonify({"success": False, "message": _TYPE_IS_FIXED}), 400)
if update_data or "actions" in data:
check_action(ra, "edit")
if "actions" in data:
# The stored actions are the schema, and a fixed value is
# the owner's (as on /api/update_tool_actions).
try:
update_data["actions"] = merge_submitted_actions(
_stored_actions(tool_doc), data["actions"], may_change_pins=ra.access == "owner",
)
except PinChangeRefused as err:
return make_response(jsonify({"success": False, "message": str(err)}), 403)
except ValueError as err:
return make_response(jsonify({"success": False, "message": str(err)}), 400)
if "config" in data:
tool_name = tool_doc.get("name", data.get("name"))
tool_name = tool_doc.get("name")
existing_config = tool_doc.get("config", {}) or {}
if tool_name == "mcp_tool":
check_oauth_mcp_owner_only(ra, existing_config, data["config"])
if _connection_server_moved(tool_doc, data["config"]):
return make_response(jsonify({"success": False, "message": _MOVE_THROUGH_MCP_SAVE}), 400)
if tool_name == "api_tool":
check_api_tool_fixed_values(ra, data["config"], existing_config)
if tool_name == "api_tool" and not _api_tool_config_needs_credentials(
data["config"], existing_config
):
@@ -1048,6 +1192,15 @@ class UpdateToolConfig(Resource):
)
@api.doc(description="Update the configuration of a tool")
def post(self):
"""Replace a tool's config, keeping stored secrets the client left out.
A connection-backed tool's new key goes to its connection (the
owner's to change) and its server cannot be moved here; an api_tool's
fixed values are the owner's.
Returns:
``{"success": true}``, or 400 / 403 / 404 with a message.
"""
decoded_token = request.decoded_token
if not decoded_token:
return make_response(jsonify({"success": False}), 401)
@@ -1088,12 +1241,18 @@ class UpdateToolConfig(Resource):
jsonify({"success": False, "message": "Invalid server URL"}),
400,
)
if _connection_server_moved(tool_doc, data["config"]):
return make_response(jsonify({"success": False, "message": _MOVE_THROUGH_MCP_SAVE}), 400)
if tool_name == "api_tool":
check_api_tool_fixed_values(ra, data["config"], tool_doc.get("config"))
tool_instance = tool_manager.tools.get(tool_name)
config_requirements = (
tool_instance.get_config_requirements() if tool_instance else {}
)
existing_config = tool_doc.get("config", {}) or {}
has_existing_secrets = "encrypted_credentials" in existing_config
has_existing_secrets = (
"encrypted_credentials" in existing_config or bool(tool_doc.get("connection_id"))
)
if config_requirements:
validation_errors = _validate_config(
@@ -1110,7 +1269,17 @@ class UpdateToolConfig(Resource):
400,
)
final_config = _prepare_tool_config(tool_doc, data["config"], config_requirements)
config = data["config"]
if tool_doc.get("connection_id"):
# The tool runs with its connection's key: a new one goes
# there (the owner's to change), not into this config.
config = _update_connection_secrets(conn, user, tool_doc, config, config_requirements)
if config is None:
return make_response(
jsonify({"success": False, "message": "Only the owner can change the credentials"}),
403,
)
final_config = _prepare_tool_config(tool_doc, config, config_requirements)
repo.update(str(tool_doc["id"]), ra.owner_id, {"config": final_config})
except AccessDenied as err:
+50
View File
@@ -304,3 +304,53 @@ class TestArgumentsShownForACall:
shown = sent_arguments(action, {"id": "1"})
assert shown == {"id": "1", "api_key": FIXED_MASK, "token": FIXED_MASK}
assert "secret" not in str(shown) and "sk-" not in str(shown)
@pytest.mark.unit
class TestStoredValuesNeverShown:
"""A value the model did not send came from the owner's stored config,
which for an api_tool query parameter is a decrypted secret."""
def test_a_stored_default_the_model_did_not_send_is_masked(self):
from docsgpt.agents.tool_pins import FIXED_MASK, sent_arguments
action = {
"query_params": {"properties": {"id": _llm(), "token": _llm(value="q-secret")}},
"body": {"properties": {"note": _llm(value="b-secret")}},
}
shown = sent_arguments(action, {"id": "1"})
assert shown == {"id": "1", "token": FIXED_MASK, "note": FIXED_MASK}
# What the model itself sent is its own and shows as it is.
assert sent_arguments(action, {"id": "1", "token": "mine"})["token"] == "mine"
def test_a_restored_api_tool_secret_is_not_recorded_with_the_call(self, mock_tool_manager, monkeypatch):
from docsgpt.api.user.tools.routes import _seal_api_tool_secrets
config = _seal_api_tool_secrets({"actions": {"get_item": {
"name": "get_item", "url": "https://api.example.com/items", "method": "GET", "active": True,
"headers": {"properties": {}},
"query_params": {"properties": {"id": _llm(), "token": _llm(value="q-secret")}},
"body": {"properties": {}},
}}}, {}, "owner")
tools_dict = {"t1": {"id": "00000000-0000-0000-0000-000000000001", "user_id": "owner",
"name": "api_tool", "config": config}}
monkeypatch.setattr(
"docsgpt.agents.tool_executor.ToolActionParser",
lambda _cls, **kw: Mock(parse_args=Mock(return_value=("t1", "get_item", {"id": "1"}))),
)
call = Mock()
call.name = "get_item"
call.id = "c1"
executor = ToolExecutor(user="u")
events = []
gen = executor.execute(tools_dict, call, "MockLLM")
while True:
try:
events.append(next(gen))
except StopIteration:
break
# The call still sends the value; the chat never shows it.
_, kwargs = mock_tool_manager.load_tool.call_args
assert kwargs["tool_config"]["query_params"]["token"] == "q-secret"
assert "q-secret" not in str(events)
assert "q-secret" not in str(executor.get_truncated_tool_calls())
+221
View File
@@ -778,3 +778,224 @@ class TestSharedOAuthMCPConfig:
body = {"id": str(tool["id"]), "config": dict(self.OTHER)}
assert _call(app, pg_conn, UpdateToolConfig, OWNER, json=body).status_code == 200
assert _row(pg_conn, tool["id"])["config"]["server_url"] == self.OTHER["server_url"]
# ---------------------------------------------------------------------------
# 7. /api/update_tool keeps fixed values and the tool type
# ---------------------------------------------------------------------------
def _pinned_tool(conn, name="ntfy"):
"""A tool whose ``server_url`` the owner fixed; ``message`` is the model's."""
return UserToolsRepository(conn).create(
OWNER, name, config={}, display_name=name, description="",
actions=[{
"name": "send", "active": True, "require_approval": False,
"parameters": {"type": "object", "properties": {
"server_url": {"type": "string", "filled_by_llm": False, "value": "https://ntfy.example.com"},
"message": {"type": "string", "filled_by_llm": True, "value": ""},
}},
}],
status=True,
)
def _moved_pin(actions):
import copy
out = copy.deepcopy(actions)
out[0]["parameters"]["properties"]["server_url"]["value"] = "https://evil.example"
return out
class TestUpdateToolFixedValues:
def test_editor_cannot_change_a_fixed_value_through_update_tool(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool = _pinned_tool(pg_conn)
_share(pg_conn, tool["id"], "ed", "editor")
body = {"id": str(tool["id"]), "actions": _moved_pin(tool["actions"])}
resp = _call(app, pg_conn, UpdateTool, "ed", json=body)
assert resp.status_code == 403
pinned = _row(pg_conn, tool["id"])["actions"][0]["parameters"]["properties"]["server_url"]
assert pinned["value"] == "https://ntfy.example.com"
def test_editor_cannot_release_a_fixed_value_through_update_tool(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool = _pinned_tool(pg_conn)
_share(pg_conn, tool["id"], "ed", "editor")
actions = tool["actions"]
actions[0]["parameters"]["properties"]["server_url"]["filled_by_llm"] = True
resp = _call(app, pg_conn, UpdateTool, "ed", json={"id": str(tool["id"]), "actions": actions})
assert resp.status_code == 403
assert _row(pg_conn, tool["id"])["actions"][0]["parameters"]["properties"]["server_url"][
"filled_by_llm"] is False
def test_editor_round_trip_with_other_edits_still_saves(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool = _pinned_tool(pg_conn)
_share(pg_conn, tool["id"], "ed", "editor")
actions = tool["actions"]
actions[0]["require_approval"] = True
body = {"id": str(tool["id"]), "name": "ntfy", "customName": "Alerts", "actions": actions}
assert _call(app, pg_conn, UpdateTool, "ed", json=body).status_code == 200
row = _row(pg_conn, tool["id"])
assert row["custom_name"] == "Alerts" and row["actions"][0]["require_approval"] is True
def test_actions_cannot_be_added_through_update_tool(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool = _pinned_tool(pg_conn)
actions = [*tool["actions"], {"name": "exfiltrate", "active": True}]
resp = _call(app, pg_conn, UpdateTool, OWNER, json={"id": str(tool["id"]), "actions": actions})
assert resp.status_code == 400
assert [a["name"] for a in _row(pg_conn, tool["id"])["actions"]] == ["send"]
def test_owner_changes_a_fixed_value_through_update_tool(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool = _pinned_tool(pg_conn)
body = {"id": str(tool["id"]), "actions": _moved_pin(tool["actions"])}
assert _call(app, pg_conn, UpdateTool, OWNER, json=body).status_code == 200
pinned = _row(pg_conn, tool["id"])["actions"][0]["parameters"]["properties"]["server_url"]
assert pinned["value"] == "https://evil.example"
@pytest.mark.parametrize("user", [OWNER, "ed"])
def test_tool_type_cannot_change(self, app, pg_conn, user):
from docsgpt.api.user.tools.routes import UpdateTool
tool = _pinned_tool(pg_conn, name="telegram")
_share(pg_conn, tool["id"], "ed", "editor")
resp = _call(app, pg_conn, UpdateTool, user, json={"id": str(tool["id"]), "name": "ntfy"})
assert resp.status_code == 400
assert _row(pg_conn, tool["id"])["name"] == "telegram"
# ---------------------------------------------------------------------------
# 8. api_tool: a stored header / query value stays fixed for non-owners
# ---------------------------------------------------------------------------
class TestApiToolFixedValues:
def _tool(self, conn):
from docsgpt.api.user.tools.routes import _seal_api_tool_secrets
tool = _tool(conn, name="api_tool", config=_seal_api_tool_secrets(_api_config(), {}, OWNER))
_share(conn, tool["id"], "ed", "editor")
return tool
def _masked(self):
return TestApiToolSecrets()._masked(_api_config())
def _token(self, cfg):
return cfg["actions"]["get_users"]["query_params"]["properties"]["token"]
@pytest.mark.parametrize("route", ["UpdateTool", "UpdateToolConfig"])
def test_editor_cannot_hand_a_stored_secret_to_the_model(self, app, pg_conn, route):
from docsgpt.api.user.tools import routes
tool = self._tool(pg_conn)
cfg = self._masked()
self._token(cfg)["filled_by_llm"] = True
resp = _call(app, pg_conn, getattr(routes, route), "ed", json={"id": str(tool["id"]), "config": cfg})
assert resp.status_code == 403
token = _runtime_action(pg_conn, tool["id"])["query_params"]["properties"]["token"]
assert token["filled_by_llm"] is False and token["value"] == "q-secret"
@pytest.mark.parametrize("change", [
{"has_value": False},
{"value": "attacker-token"},
])
def test_editor_cannot_clear_or_replace_a_stored_value(self, app, pg_conn, change):
from docsgpt.api.user.tools.routes import UpdateTool
tool = self._tool(pg_conn)
cfg = self._masked()
self._token(cfg).update(change)
resp = _call(app, pg_conn, UpdateTool, "ed", json={"id": str(tool["id"]), "config": cfg})
assert resp.status_code == 403
assert _runtime_action(pg_conn, tool["id"])["query_params"]["properties"]["token"]["value"] == "q-secret"
def test_editor_cannot_drop_a_fixed_parameter_then_re_add_it(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool = self._tool(pg_conn)
cfg = self._masked()
del cfg["actions"]["get_users"]["query_params"]["properties"]["token"]
resp = _call(app, pg_conn, UpdateTool, "ed", json={"id": str(tool["id"]), "config": cfg})
assert resp.status_code == 403
def test_editor_still_adds_an_empty_parameter_and_edits_descriptions(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool = self._tool(pg_conn)
cfg = self._masked()
cfg["actions"]["get_users"]["description"] = "Edited"
cfg["actions"]["get_users"]["query_params"]["properties"]["page"] = {
"type": "string", "description": "", "value": "", "filled_by_llm": False, "required": False,
}
resp = _call(app, pg_conn, UpdateTool, "ed", json={"id": str(tool["id"]), "config": cfg})
assert resp.status_code == 200, resp.json
action = _runtime_action(pg_conn, tool["id"])
assert action["description"] == "Edited"
assert action["query_params"]["properties"]["token"]["value"] == "q-secret"
def test_owner_may_hand_a_value_to_the_model(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool = self._tool(pg_conn)
cfg = self._masked()
self._token(cfg)["filled_by_llm"] = True
assert _call(app, pg_conn, UpdateTool, OWNER, json={"id": str(tool["id"]), "config": cfg}).status_code == 200
assert _runtime_action(pg_conn, tool["id"])["query_params"]["properties"]["token"]["filled_by_llm"] is True
# ---------------------------------------------------------------------------
# 9. A connection-backed MCP tool moves only through /api/mcp_server/save
# ---------------------------------------------------------------------------
class TestConnectionBackedMCPConfig:
SAME = {"server_url": "https://mcp.example.com/mcp", "auth_type": "bearer", "transport_type": "http"}
@pytest.mark.parametrize("route", ["UpdateTool", "UpdateToolConfig"])
@pytest.mark.parametrize("change", [
{"server_url": "https://other.example.org/mcp", "bearer_token": "new"},
{"server_url": "http://mcp.example.com/mcp", "bearer_token": "new"},
{"auth_type": "api_key", "api_key": "new"},
])
def test_moving_the_server_or_auth_is_refused(self, app, pg_conn, route, change):
from docsgpt.api.user.tools import routes
tool, connection = _mcp_connection_tool(pg_conn, {"bearer_token": "tok"})
body = {"id": str(tool["id"]), "config": {**self.SAME, **change}}
resp = _call(app, pg_conn, getattr(routes, route), OWNER, json=body)
assert resp.status_code == 400
assert "mcp_server/save" in resp.json["message"]
row = _row(pg_conn, tool["id"])
assert row["config"]["server_url"] == self.SAME["server_url"]
assert row["config"]["auth_type"] == "bearer"
assert str(row["connection_id"]) == str(connection["id"])
def test_update_tool_config_writes_a_new_key_to_the_connection(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateToolConfig
tool, _connection = _mcp_connection_tool(pg_conn, {"bearer_token": "tok"})
body = {"id": str(tool["id"]), "config": {**self.SAME, "bearer_token": "tok2"}}
assert _call(app, pg_conn, UpdateToolConfig, OWNER, json=body).status_code == 200
assert _stored_mcp_secret(pg_conn, tool["id"]) == ({"bearer_token": "tok2"}, OWNER)
def test_update_tool_config_keeps_the_owners_connection_from_editors(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateToolConfig
tool, _connection = _mcp_connection_tool(pg_conn, {"bearer_token": "tok"})
_share(pg_conn, tool["id"], "ed", "editor")
body = {"id": str(tool["id"]), "config": {**self.SAME, "bearer_token": "tok2"}}
assert _call(app, pg_conn, UpdateToolConfig, "ed", json=body).status_code == 403
assert _stored_mcp_secret(pg_conn, tool["id"]) == ({"bearer_token": "tok"}, OWNER)
def test_same_server_other_path_still_saves(self, app, pg_conn):
from docsgpt.api.user.tools.routes import UpdateTool
tool, connection = _mcp_connection_tool(pg_conn, {"bearer_token": "tok"})
body = {"id": str(tool["id"]), "config": {**self.SAME, "server_url": "https://mcp.example.com/v2/mcp"}}
assert _call(app, pg_conn, UpdateTool, OWNER, json=body).status_code == 200
row = _row(pg_conn, tool["id"])
assert row["config"]["server_url"] == "https://mcp.example.com/v2/mcp"
assert str(row["connection_id"]) == str(connection["id"])