This is an automated email from the ASF dual-hosted git repository.
CRZbulabula pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iotdb-mcp-server.git
The following commit(s) were added to refs/heads/main by this push:
new 4550cff fix: restrict agent permission (#20)
4550cff is described below
commit 4550cffb51cfb5ea3a98a81000ac5ce075e61ee0
Author: Zeyu Zhang <[email protected]>
AuthorDate: Thu Sep 17 15:47:21 2026 +0800
fix: restrict agent permission (#20)
---
README-zh.md | 7 +
README.md | 9 +
docs/iotdb-threat-model-proposal.md | 95 ++++++
docs/session-policy-security.md | 111 +++++++
pyproject.toml | 1 +
src/iotdb_mcp_server/policy_approval.py | 171 ++++++++++
src/iotdb_mcp_server/runtime_policy.py | 288 +++++++++++++----
src/iotdb_mcp_server/server.py | 2 +
src/iotdb_mcp_server/services/README.md | 38 ++-
src/iotdb_mcp_server/services/database.py | 2 +-
src/iotdb_mcp_server/services/metadata.py | 2 +-
src/iotdb_mcp_server/services/model.py | 2 +-
src/iotdb_mcp_server/services/runtime_policy.py | 27 +-
src/iotdb_mcp_server/services/sql_driver.py | 5 +-
src/iotdb_mcp_server/services/table.py | 2 +-
src/iotdb_mcp_server/services/timeseries.py | 2 +-
src/iotdb_mcp_server/services/ttl.py | 2 +-
src/iotdb_mcp_server/services/write.py | 2 +-
tests/test_runtime_policy.py | 414 +++++++++++++++++++++++-
19 files changed, 1076 insertions(+), 106 deletions(-)
diff --git a/README-zh.md b/README-zh.md
index 112d2f5..fc98fb4 100644
--- a/README-zh.md
+++ b/README-zh.md
@@ -27,6 +27,13 @@ IoTDB MCP 权限默认是提示层。服务器为 SQL 操作返回所需权限
只有需要 MCP server 自身硬阻断时，才设置
`TIMESEEK_MCP_PERMISSION_ENFORCEMENT=strict`。
+Session 策略受启动时冻结的部署权限上限约束（进程环境变量 > MCP 配置 > 默认值）。
+收紧权限立即生效；在上限内扩大权限默认需要独立的管理员审批，审批前保持原策略。
+部署者可设置 `IOTDB_SESSION_POLICY_APPROVAL_MODE=allow`，允许上限内免审批调整。
+普通工具不能修改强制检查开关、SQL 分类扩展项或突破部署上限；reset 和 replace
+也遵循相同规则。配置与审批步骤见[Session 权限管理](docs/session-policy-security.md)。
+审批目录和管理员命令必须处于 agent 的文件系统及 shell 权限之外。
+
### 工具
服务器为 IoTDB 的树模型（Tree Model）和表模型（Table Model）提供了不同的工具。您可以通过设置
"IOTDB_SQL_DIALECT" 配置为 "tree" 或 "table" 来选择使用哪种模型。
diff --git a/README.md b/README.md
index 622e785..1b2a133 100644
--- a/README.md
+++ b/README.md
@@ -29,6 +29,15 @@ environment variables such as `IOTDB_SQL_DRIVER_MODE=full`
and
`TIMESEEK_MCP_PERMISSION_ENFORCEMENT=strict` only when the MCP server itself
should hard-block disallowed operations.
+Session policy changes are bounded by a deployment policy frozen at startup
+(process environment > MCP configuration > defaults). Narrowing applies
immediately;
+widening within that ceiling requires out-of-band administrator approval by
default.
+`IOTDB_SESSION_POLICY_APPROVAL_MODE=allow` explicitly permits in-ceiling
widening
+without approval. Neither setting allows tools to change enforcement switches,
+SQL classification prefixes, or the deployment ceiling. Reset and replace
follow
+the same rules. See [session policy
administration](docs/session-policy-security.md)
+for configuration, approval commands, and the required host isolation.
+
### Tools
The server offers different tools for IoTDB Tree Model and Table Model. You
can choose between them by setting the "IOTDB_SQL_DIALECT" configuration to
either "tree" or "table".
diff --git a/docs/iotdb-threat-model-proposal.md
b/docs/iotdb-threat-model-proposal.md
new file mode 100644
index 0000000..c467969
--- /dev/null
+++ b/docs/iotdb-threat-model-proposal.md
@@ -0,0 +1,95 @@
+# IoTDB 威胁模型补充建议：MCP Server
+
+本文件是提交 PMC 审阅的建议，尚不是已批准的安全承诺。
+按本地 IoTDB 主仓库 `THREAT_MODEL.md` 的 v0 章节组织；合入前需与目标版本核对。
+当前 MCP 的运行约束见 [Session 权限管理](session-policy-security.md)。
+
+## §2：明确组件与参与者
+
+在组件表中显式增加 `apache/iotdb-mcp-server`，由 PMC 确认纳入主模型还是维护单独模型并互相引用。
+不要因其位于独立仓库，就默认套用客户端 SDK 的范围排除。
+
+建议组件表条目：
+
+| Family | Entry point | Touches outside process | In model? |
+| --- | --- | --- | --- |
+| IoTDB MCP Server | MCP tools over stdio | IoTDB session RPC, local
configuration, result files, policy approval files | In the MCP-specific
boundary described below; database-side RBAC remains independently enforced |
+
+明确四种参与者：部署管理员、负责交互与沙箱的宿主、调用 MCP 工具的 agent/MCP 客户端、
+数据库账号。宿主可以管理 agent 的执行权限，但 agent 不因此自动获得管理员身份。
+数据库凭据由 MCP 使用时，数据库账号权限也不等于 agent 对 MCP 管理策略的权限。
+
+## §3、§4、§7、§11a：修正 root/admin 排除规则
+
+现有模型把“需要 root/admin 的操作”视为可信输入。对 MCP 应补充例外：
+
+> An MCP client does not become a trusted deployment administrator merely
because
+> the MCP server uses a database account with administrative privileges. An MCP
+> client bypassing an operator-enforced MCP policy is an in-scope adversary
even
+> when the underlying database would authorize the SQL for the server's
account.
+
+在威胁主体中增加：已连接且能调用普通 MCP 工具的恶意客户端、被提示注入影响的 agent、
+超出用户意图自行执行操作的 agent。攻击者可选择任意工具名、参数、调用顺序和并发请求，
+也可自行填写危险操作确认布尔值。不能假设工具调用来自自然语言用户的真实授权。
+
+## §4：增加两条边界
+
+1. **agent → MCP 数据操作与 session 策略工具**：不可信输入；部署策略上限在服务启动时冻结。
+ 显式越权请求不可修改策略；收紧可直接生效；上限内扩大按部署者选定的审批模式执行。
+2. **管理员 → 配置与审批通道**：独立的可信管理操作；修改上限、enforcement 和 SQL 分类规则
+ 需要受保护的配置及重启。普通 MCP 工具不能批准自己的请求。
+
+数据操作还要跨越 **MCP → IoTDB RBAC** 边界。MCP 限制与数据库授权应同时满足，
+MCP 的部署上限不能扩大数据库账号本身的权限。
+
+## §5、§5a、§6：部署假设与输入分类
+
+- stdio 仅说明传输方式，不说明客户端可信；一个进程是一个策略域，不提供多租户隔离。
+- 管理配置、审批目录和服务代码必须不在 agent 可读写或可替换的权限范围内；
+ 受信管理端可以访问它们。不得将审批目录放在可由查询导出等工具写入的位置。
+- 同用户不受限 shell 能运行管理员命令或伪造审批文件。0700 不是同 UID 的安全隔离；
+ 若需要约束该 agent，宿主必须提供额外隔离。管理员自动批准服务也属于可信计算基。
+- 审批模式 `require`（默认）和 `allow`（管理员显式选择）必须列为安全配置变体。
+ require 无可用通道时禁止扩大，不可自动降级为 allow。
+- 明确 `advisory` 不提供 SQL 权限强制执行保证；`strict` 下才承诺相关执行检查。
+ 当前默认 advisory 应清楚记录，PMC 可另行决定未来版本是否切换默认 strict。
+- `confirm_destructive=true` 是攻击者可控输入，不等于人类身份、签名或独立审批凭据。
+- 审批请求内容与决定是不同信任级别的输入。审批需绑定确切变更及有效期，防止替换与重放。
+
+## §8：建议承诺的 MCP 安全属性
+
+建议 PMC 明确承诺并逐个核验调用入口：
+
+1. **部署上限不可由工具扩大**：set、preset、replace、reset、工具别名、并发与重启恢复
+ 均不得绕过。enforcement 和 SQL 分类规则不能由普通工具改写。
+2. **先审批后生效**：require 模式下，未批准、拒绝、过期、不匹配或已消费的审批均不能扩大权限。
+ 对混合权限变更，拒绝或待批时不得部分生效。
+3. **审批绑定**：批准仅用于特定运行实例、策略修订和具体变更，不可挪用到另一提权请求。
+4. **严格执行一致性**：strict 模式的各 SQL/DDL/DML 工具必须按其声明的权限规则检查。
+ 主模型应说明通用 SQL driver 与专用工具开关之间的关系，避免部署者误认为关闭单个入口
+ 等于关闭所有等价 SQL。实现完整入口审计后再扩大此项承诺。
+5. **凭据保密与资源生命周期**：建议分别定义快照/日志不泄露认证凭据、结果文件路径约束、
+ 查询与连接池资源释放的保证及边界，并以独立测试核验；本次策略修改不代表已完成这些审计。
+
+## §9、§10、§11：非保证与部署者职责
+
+- 不防御已控制 MCP 进程、服务代码、管理配置或审批文件的攻击者；但这种控制不能从
+ “能够调用 MCP 工具”或“服务器使用 root 数据库凭据”直接推定。
+- 不把自然语言“用户已同意”、普通确认布尔值或 agent full permission 当成人工审批证明。
+- 不宣称 advisory 模式强制只读，也不宣称 session 策略收紧会撤销已经开始执行的 SQL。
+- 部署者负责 strict 配置、IoTDB 最小权限账号、管理通道隔离、审批目录保护及实例隔离。
+ 如果显式选择 allow，仅表示同意部署上限内自动调整，不表示同意修改部署上限或关闭 strict。
+
+## §12、§13：模型变更与分流示例
+
+| 场景 | 建议分流 |
+| --- | --- |
+| 普通 MCP 客户端能关闭已启用的 strict 或突破部署上限 | VALID：权限边界绕过 |
+| require 模式下可伪造、替换或重放审批而扩大权限 | VALID：审批边界绕过 |
+| 使用部署者明确开启的 allow 在部署上限内扩大 session 权限 | BY-DESIGN，前提是配置契约清楚 |
+| 仅证明显式 advisory 不强制执行 SQL 权限 | 按已声明非保证判断；误导文档另行评估 |
+| 已掌握管理员 shell 后编辑配置重启 | OUT-OF-MODEL: trusted-input，须证明已有管理权限 |
+| 服务端持有高权限数据库账号，而 agent 仅有普通 MCP 工具权限 | 不得仅凭账号级别判为 trusted-input |
+
+新增 HTTP/共享服务传输、多客户端共享实例、在线管理 API、宿主自动审批集成，或修改默认
+enforcement/审批模式时，应触发模型复审。最终范围和承诺由 PMC 确认，并随发布版本绑定。
diff --git a/docs/session-policy-security.md b/docs/session-policy-security.md
new file mode 100644
index 0000000..a631682
--- /dev/null
+++ b/docs/session-policy-security.md
@@ -0,0 +1,111 @@
+# Session 权限管理
+
+## 权限边界
+
+服务启动时冻结部署策略：进程环境变量优先，其次为启动时发现的 MCP 配置，最后为默认值。
+运行期间修改环境变量或 `.mcp.json` 不会改变该策略。修改部署策略后须由管理员重启服务。
+部署上限不是启动 session 的只读默认值；需要“部署允许写、session 暂时只读”时，
+先配置可写部署上限，再通过 session 工具收紧。
+
+策略比较规则：
+
+| 类型 | 收紧 | 扩大 |
+| --- | --- | --- |
+| `IOTDB_ENABLE_*` | true → false | false → true |
+| `*_ALLOWED_USERS` | 删除用户，`*` → 指定集合 | 添加用户，指定集合 → `*` |
+| `IOTDB_SQL_DRIVER_MODE` | full → ddl → readonly | readonly → ddl → full |
+| `*CONFIRM*` | false → true | true → false |
+
+空用户白名单表示拒绝所有用户。`readonly` 预设也关闭模型管理。
+预设与部署上限取交集，例如部署只允许 alice 时，`full` 不会把用户白名单改成 `*`。
+显式 `policy` 中任何一项超过部署上限，整个请求都会被拒绝；不会应用其余项。
+混合收紧和扩大的请求作为整体审批，在批准之前不应用其中任何一项。
+
+以下内容仅由管理员配置，普通 MCP 工具不能设置或 reset：
+
+- `IOTDB_STRICT_PERMISSION_ENFORCEMENT`
+- `TIMESEEK_MCP_PERMISSION_ENFORCEMENT`
+- `IOTDB_SQL_DRIVER_EXTRA_READONLY_PREFIXES`
+- `IOTDB_SQL_DRIVER_EXTRA_DDL_PREFIXES`
+- `IOTDB_SQL_DRIVER_EXTRA_FULL_PREFIXES`
+- 审批模式和审批目录
+
+SQL 分类扩展项可能把写语句划为只读，因此也不能交给 agent 修改。
+关闭严格检查、扩大部署上限或修改 SQL 分类规则的管理员通道是：
+在受保护的宿主配置中修改相应项，然后重启 MCP。session 审批命令不具备这些能力。
+
+## 上限内提权审批
+
+`IOTDB_SESSION_POLICY_APPROVAL_MODE` 由管理员在启动时设置：
+
+- `require`（默认）：收紧立即生效；扩大需由独立管理员通道批准。
+- `allow`：允许上限内扩大权限，无需逐次审批。仍不能越过部署上限或修改管理员专属项。
+
+`set_iotdb_session_policy`、`replace=true`、`reset_iotdb_session_policy` 均遵循此规则。
+重置只读 session 通常会恢复更宽权限，所以也需要审批。
+未配置审批目录时，扩大请求返回 `approval_unavailable`、`applied=false`，原权限保持不变。
+无变化的请求不需要审批。
+
+管理员在 agent 沙箱之外创建目录，并在 MCP 启动环境中配置：
+
+```sh
+mkdir -m 700 /absolute/protected/iotdb-policy-approvals
+export
IOTDB_SESSION_POLICY_APPROVAL_DIR=/absolute/protected/iotdb-policy-approvals
+export IOTDB_SESSION_POLICY_APPROVAL_MODE=require
+export TIMESEEK_MCP_PERMISSION_ENFORCEMENT=strict
+```
+
+目录需由服务运行用户拥有；POSIX 上禁止 group/other 权限。管理员命令在可信终端中
+以能访问该目录的服务用户身份运行。Windows 部署需用 ACL 和宿主沙箱落实等效隔离。
+不要把目录设在 agent 可写工作区、SQL 结果导出目录或其他工具可写路径中。
+
+1. agent 请求扩大 session 权限。返回 `status=approval_required`、`applied=false`、
+ `request_id`、变更前后值和过期时间；此时策略尚未改变。
+2. 管理员从可信界面查看请求：
+
+ ```sh
+ iotdb-policy-admin show REQUEST_ID --directory
/absolute/protected/iotdb-policy-approvals
+ ```
+
+3. 核对具体变更后，从独立管理终端批准或拒绝：
+
+ ```sh
+ iotdb-policy-admin approve REQUEST_ID --directory
/absolute/protected/iotdb-policy-approvals
+ # 或：iotdb-policy-admin deny REQUEST_ID --directory
/absolute/protected/iotdb-policy-approvals
+ ```
+
+4. agent 重试原始 MCP 调用。服务校验对应审批后才应用变更，返回 `applied=true`。
+
+源码运行可用 `PYTHONPATH=src python3 -m iotdb_mcp_server.policy_approval` 替代命令名。
+管理命令不会注册为 MCP tool；MCP 调用没有 `approved` 或 `confirm` 提权参数。
+批准绑定具体变更、部署指纹、策略修订号和运行实例；有效期 5 分钟，仅可消费一次。
+策略发生变化、服务重启或请求过期后，旧审批不能用于新的变更。最多保留 32 个待审批请求；
+在新申请时清理本实例已过期或旧修订请求。重启遗留文件不会被新实例采纳，可由管理员清理。
+审批文件是授权凭据，不是不可篡改审计日志；生产环境可由管理平台另行记录审批者身份。
+
+## 与 agent full permission 的关系
+
+宿主的 full permission 不会自动改变服务端审批模式。人工审批流程应由宿主在独立权限域
+执行，不应把上述审批命令直接交给 agent 自动运行。
+
+**同一 OS 用户下，若 agent 具有未受限制的 shell/文件访问，0700 不能隔离它与 MCP。**
+部署者必须通过沙箱、容器、独立服务身份或可信管理平台，阻止 agent 读写审批目录、
+修改部署配置、替换服务代码或自行重启为更宽权限。若这些能力已授权给 agent，
+它拥有管理员通道的能力，服务端不能再保证人工审批。普通工具自动批准不能代替此授权。
+
+## 保证范围与兼容性
+
+- 默认 SQL 检查仍为 `advisory`；要让只读、工具开关和用户白名单成为硬限制，管理员必须启用
+ `TIMESEEK_MCP_PERMISSION_ENFORCEMENT=strict` 或
`IOTDB_STRICT_PERMISSION_ENFORCEMENT=true`。
+ 在 advisory 模式下，session 配置收紧不会提供 SQL 执行隔离保证。
+- SQL 执行工具的 `confirm_destructive` 等布尔值仍是调用者声明，不是可验证的人工审批。
+ 本变更的独立审批针对 session 权限扩大，不是每条危险 SQL 的审批服务。
+- 通用 SQL driver 的模式/白名单与专用 DDL/DML 工具开关分别控制各自入口。
+ 部署只读时应同时设置 driver 为 readonly 并关闭相关专用写工具；可用 readonly 预设收紧
+ session，但其部署上限仍以管理员配置为准。IoTDB 账号权限继续独立生效，建议最小授权。
+- 策略是单个 stdio MCP 进程级别的，不是多租户或多目标独立授权域。
+ 不应把此进程在不同信任级别的客户端间共享；正在执行的 SQL 不承诺被策略收紧立即取消。
+- 相较旧版，配置不再热加载，进程环境变量优先，非法策略值会使初始化失败；
+ reset/replace 可能返回待审批而非立即完成。调用方应检查 `payload.applied` 和 `payload.status`。
+
+IoTDB 主仓库的威胁模型补充草案见
[iotdb-threat-model-proposal.md](iotdb-threat-model-proposal.md)。
diff --git a/pyproject.toml b/pyproject.toml
index ca73270..65d3c8a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -34,3 +34,4 @@ dev-dependencies = [
[project.scripts]
iotdb-mcp-server = "iotdb_mcp_server:main"
iotdb-target-cli = "iotdb_mcp_server.target_cli:main"
+iotdb-policy-admin = "iotdb_mcp_server.policy_approval:main"
diff --git a/src/iotdb_mcp_server/policy_approval.py
b/src/iotdb_mcp_server/policy_approval.py
new file mode 100644
index 0000000..0f08fe9
--- /dev/null
+++ b/src/iotdb_mcp_server/policy_approval.py
@@ -0,0 +1,171 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Out-of-band policy approvals. Never register administrative functions as
tools.
+
+The directory and this CLI must be outside the agent's filesystem/shell
authority.
+Filesystem permissions alone do not isolate an unsandboxed same-UID agent.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import tempfile
+import time
+from typing import Any
+import uuid
+
+_APPROVAL_TTL_SECONDS = 300
+_MAX_PENDING = 32
+
+
+def _digest(value: dict[str, Any]) -> str:
+ return hashlib.sha256(json.dumps(value,
sort_keys=True).encode()).hexdigest()
+
+
+def _check_directory(directory: Path) -> None:
+ if not directory.is_absolute() or not directory.is_dir() or
directory.is_symlink():
+ raise ValueError(
+ "Approval directory must be an existing, absolute, non-symlink
directory."
+ )
+ if os.name == "posix":
+ info = directory.stat()
+ if info.st_uid != os.getuid() or info.st_mode & 0o077:
+ raise PermissionError(
+ "Approval directory must be owned by the service user with
mode 0700."
+ )
+
+
+def _write_json(path: Path, data: dict[str, Any]) -> None:
+ fd, name = tempfile.mkstemp(prefix=".policy-", dir=path.parent)
+ try:
+ with os.fdopen(fd, "w", encoding="utf-8") as stream:
+ json.dump(data, stream, indent=2, sort_keys=True)
+ stream.write("\n")
+ os.replace(name, path)
+ finally:
+ Path(name).unlink(missing_ok=True)
+
+
+class PolicyApprovalStore:
+ """Bind one decision to an exact proposal, revision, and running
process."""
+
+ def __init__(self, directory: Path):
+ _check_directory(directory)
+ self.directory = directory
+ self._pending: dict[str, dict[str, Any]] = {}
+
+ def _remove(self, key: str) -> None:
+ request = self._pending.pop(key)
+ for suffix in ("request", "decision"):
+ (self.directory / f"{request['request_id']}.{suffix}.json").unlink(
+ missing_ok=True
+ )
+
+ def authorize(self, proposal: dict[str, Any]) -> dict[str, Any]:
+ _check_directory(self.directory)
+ now = time.time()
+ for key, request in list(self._pending.items()):
+ if (
+ request["expires_at"] <= now
+ or request["revision"] != proposal["revision"]
+ ):
+ self._remove(key)
+ key = _digest(proposal)
+ if key not in self._pending:
+ if len(self._pending) >= _MAX_PENDING:
+ raise PermissionError(
+ "Too many pending policy approvals. Wait for requests to
expire."
+ )
+ request = {
+ **proposal,
+ "request_id": uuid.uuid4().hex,
+ "created_at": now,
+ "expires_at": now + _APPROVAL_TTL_SECONDS,
+ }
+ _write_json(
+ self.directory / f"{request['request_id']}.request.json",
request
+ )
+ self._pending[key] = request
+ request = self._pending[key]
+ result = {
+ "status": "approval_required",
+ "request_id": request["request_id"],
+ "expires_at": request["expires_at"],
+ "requested_changes": request["changes"],
+ }
+ decision_path = self.directory /
f"{request['request_id']}.decision.json"
+ try:
+ decision = json.loads(decision_path.read_text(encoding="utf-8"))
+ except FileNotFoundError:
+ return result
+ # The in-memory request, not the file shown to the administrator, is
+ # authoritative. Tampering with that file cannot approve another
change.
+ if not isinstance(decision, dict) or decision.get("request_digest") !=
_digest(
+ request
+ ):
+ raise PermissionError("Invalid administrator decision; policy
unchanged.")
+ if decision.get("decision") == "approve":
+ if time.time() >= request["expires_at"]:
+ self._remove(key)
+ return self.authorize(proposal)
+ self._remove(key)
+ result["status"] = "approved"
+ else:
+ result["status"] = "approval_rejected"
+ return result
+
+
+def main(argv: list[str] | None = None) -> int:
+ """Run only from an operator-controlled terminal or trusted management
UI."""
+ parser = argparse.ArgumentParser(
+ description="Review and decide an MCP session policy request."
+ )
+ parser.add_argument("action", choices=("show", "approve", "deny"))
+ parser.add_argument("request_id")
+ parser.add_argument("--directory", required=True, type=Path)
+ args = parser.parse_args(argv)
+ _check_directory(args.directory)
+ if not re.fullmatch(r"[0-9a-f]{32}", args.request_id):
+ parser.error("Invalid request ID")
+ request = json.loads(
+ (args.directory /
f"{args.request_id}.request.json").read_text(encoding="utf-8")
+ )
+ if request["request_id"] != args.request_id or request["expires_at"] <=
time.time():
+ parser.error("Request is invalid or expired")
+ print(json.dumps(request, indent=2, sort_keys=True))
+ if args.action != "show":
+ _write_json(
+ args.directory / f"{args.request_id}.decision.json",
+ {
+ "request_digest": _digest(request),
+ "decision": args.action,
+ },
+ )
+ print(
+ f"Decision recorded: {args.action}. Retry the original MCP request
to apply it."
+ )
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/src/iotdb_mcp_server/runtime_policy.py
b/src/iotdb_mcp_server/runtime_policy.py
index 4b3f418..b591a77 100644
--- a/src/iotdb_mcp_server/runtime_policy.py
+++ b/src/iotdb_mcp_server/runtime_policy.py
@@ -19,6 +19,7 @@
from __future__ import annotations
from collections.abc import Mapping
+import hashlib
import json
import os
from pathlib import Path
@@ -26,13 +27,23 @@ import re
from threading import RLock
from typing import Any
+from iotdb_mcp_server.policy_approval import PolicyApprovalStore
+
_DEFAULT_SERVER_NAME = "iotdb"
_SHELL_DEFAULT_RE = re.compile(r"^\$\{([A-Za-z_][A-Za-z0-9_]*):-([^}]*)\}$")
-_SENSITIVE_KEY_RE = re.compile(r"(PASSWORD|SECRET|TOKEN|API_KEY|ACCESS_KEY)",
re.IGNORECASE)
+_SENSITIVE_KEY_RE = re.compile(
+ r"(PASSWORD|SECRET|TOKEN|API_KEY|ACCESS_KEY)", re.IGNORECASE
+)
_WITHHELD_ENV_KEYS = frozenset({"TIMESEEK_IOTDB_TARGETS_JSON"})
_SESSION_POLICY_LOCK = RLock()
_SESSION_POLICY: dict[str, str] = {}
+_DEPLOYMENT_POLICY: dict[str, str] | None = None
+_DEPLOYMENT_ENV: dict[str, str] = {}
+_DEPLOYMENT_CONFIG_PATH = ""
+_APPROVAL_MODE = "require"
+_APPROVAL_STORE: PolicyApprovalStore | None = None
+_POLICY_REVISION = 0
_ENFORCEMENT_POLICY_KEYS = frozenset(
{
@@ -75,6 +86,19 @@ _POLICY_KEYS = frozenset(
}
)
+# Prefix changes can reclassify writes as reads, so they are
administrator-only.
+_ADMIN_POLICY_KEYS = _ENFORCEMENT_POLICY_KEYS | frozenset(
+ key for key in _POLICY_KEYS if "_EXTRA_" in key
+)
+_BOOL_POLICY_KEYS = frozenset(
+ key
+ for key in _POLICY_KEYS
+ if key.startswith("IOTDB_ENABLE_")
+ or "CONFIRM" in key
+ or key == "IOTDB_STRICT_PERMISSION_ENFORCEMENT"
+)
+_MODE_RANK = {"readonly": 0, "ddl": 1, "full": 2}
+
_FULL_PERMISSION_DEFAULTS = {
"IOTDB_ENABLE_METADATA_QUERY": "true",
"IOTDB_METADATA_ALLOWED_USERS": "*",
@@ -121,6 +145,7 @@ _PRESET_POLICIES = {
"IOTDB_ENABLE_TIMESERIES_DDL": "false",
"IOTDB_ENABLE_TTL_SQL": "false",
"IOTDB_ENABLE_WRITE_DML": "false",
+ "IOTDB_ENABLE_MODEL_MANAGEMENT": "false",
"IOTDB_SQL_DRIVER_MODE": "readonly",
},
}
@@ -197,16 +222,74 @@ def _load_mcp_env() -> tuple[dict[str, str], str]:
server = data.get("mcpServers", {}).get(server_name, {})
env = server.get("env", {})
if isinstance(env, dict):
- return {str(key): _expand_config_value(value) for key, value in
env.items()}, str(path)
+ return {
+ str(key): _expand_config_value(value) for key, value in
env.items()
+ }, str(path)
return {}, ""
-def dynamic_getenv(name: str, default: str | None = None) -> str | None:
+def _normalize_policy_value(key: str, value: Any) -> str:
+ text = _coerce_policy_value(value).strip()
+ if key in _BOOL_POLICY_KEYS:
+ if text.lower() in {"1", "true", "yes", "on"}:
+ return "true"
+ if text.lower() in {"0", "false", "no", "off"}:
+ return "false"
+ raise ValueError(f"{key} must be a boolean.")
+ if key == "IOTDB_SQL_DRIVER_MODE":
+ if text.lower() not in _MODE_RANK:
+ raise ValueError(f"{key} must be readonly, ddl, or full.")
+ return text.lower()
+ if key == "TIMESEEK_MCP_PERMISSION_ENFORCEMENT":
+ if text.lower() not in {"advisory", "strict"}:
+ raise ValueError(f"{key} must be advisory or strict.")
+ return text.lower()
+ items = {item.strip() for item in text.split(",") if item.strip()}
+ if key.endswith("_ALLOWED_USERS"):
+ return "*" if "*" in items else ",".join(sorted(items))
+ return ",".join(sorted(item.upper() for item in items))
+
+
+def initialize_runtime_policy() -> None:
+ """Freeze operator policy once, before tools are registered. No MCP reload
API."""
+ global _DEPLOYMENT_POLICY, _DEPLOYMENT_ENV, _DEPLOYMENT_CONFIG_PATH
+ global _APPROVAL_MODE, _APPROVAL_STORE
with _SESSION_POLICY_LOCK:
- value = _SESSION_POLICY.get(name)
- if value not in (None, ""):
- return value
+ if _DEPLOYMENT_POLICY is not None:
+ return
+ env, path = _load_mcp_env()
+ # Explicit process environment wins over config discovery, including
empty
+ # allowlists. Malformed policy fails startup rather than enabling
access.
+ source = {**env, **os.environ}
+ defaults = {**_FULL_PERMISSION_DEFAULTS, **_ENFORCEMENT_DEFAULTS}
+ deployment = {
+ key: _normalize_policy_value(key, source.get(key,
defaults.get(key, "")))
+ for key in _POLICY_KEYS
+ }
+ mode = (
+ source.get("IOTDB_SESSION_POLICY_APPROVAL_MODE",
"require").strip().lower()
+ )
+ if mode not in {"require", "allow"}:
+ raise ValueError(
+ "IOTDB_SESSION_POLICY_APPROVAL_MODE must be require or allow."
+ )
+ directory = source.get("IOTDB_SESSION_POLICY_APPROVAL_DIR", "")
+ store = PolicyApprovalStore(Path(directory).expanduser()) if directory
else None
+ _DEPLOYMENT_ENV = env
+ _DEPLOYMENT_CONFIG_PATH = path
+ _APPROVAL_MODE = mode
+ _APPROVAL_STORE = store
+ _DEPLOYMENT_POLICY = deployment
+
+def dynamic_getenv(name: str, default: str | None = None) -> str | None:
+ if name in _POLICY_KEYS:
+ initialize_runtime_policy()
+ with _SESSION_POLICY_LOCK:
+ assert _DEPLOYMENT_POLICY is not None
+ return _SESSION_POLICY.get(name, _DEPLOYMENT_POLICY[name])
+
+ # Non-policy operational settings retain their existing lookup behavior.
env, _ = _load_mcp_env()
value = env.get(name)
if value not in (None, ""):
@@ -237,7 +320,14 @@ def dynamic_env_bool(name: str, default: bool) -> bool:
def permission_enforcement_mode() -> str:
if dynamic_env_bool("IOTDB_STRICT_PERMISSION_ENFORCEMENT", False):
return "strict"
- mode = (dynamic_getenv("TIMESEEK_MCP_PERMISSION_ENFORCEMENT", "advisory")
or "advisory").strip().lower()
+ mode = (
+ (
+ dynamic_getenv("TIMESEEK_MCP_PERMISSION_ENFORCEMENT", "advisory")
+ or "advisory"
+ )
+ .strip()
+ .lower()
+ )
if mode not in {"advisory", "strict"}:
return "advisory"
return mode
@@ -249,18 +339,15 @@ def strict_permission_enforcement() -> bool:
def _redact_env(env: dict[str, str]) -> dict[str, str]:
return {
- key: "***"
- if key in _WITHHELD_ENV_KEYS or _SENSITIVE_KEY_RE.search(key)
- else value
+ key: (
+ "***"
+ if key in _WITHHELD_ENV_KEYS or _SENSITIVE_KEY_RE.search(key)
+ else value
+ )
for key, value in env.items()
}
-def _session_policy_copy() -> dict[str, str]:
- with _SESSION_POLICY_LOCK:
- return dict(_SESSION_POLICY)
-
-
def _coerce_policy_value(value: Any) -> str:
if isinstance(value, bool):
return "true" if value else "false"
@@ -275,6 +362,13 @@ def _validate_policy_keys(policy: Mapping[str, Any]) ->
None:
+ ", ".join(unknown)
+ ". Use get_iotdb_session_policy to inspect supported keys."
)
+ admin = sorted(str(key) for key in policy if str(key) in
_ADMIN_POLICY_KEYS)
+ if admin:
+ raise PermissionError(
+ "Administrator-only policy key(s): "
+ + ", ".join(admin)
+ + ". Change protected deployment configuration and restart the
server."
+ )
def session_policy_for_preset(preset: str) -> dict[str, str]:
@@ -282,11 +376,86 @@ def session_policy_for_preset(preset: str) -> dict[str,
str]:
try:
return dict(_PRESET_POLICIES[normalized])
except KeyError as exc:
- raise ValueError("Unsupported policy preset. Expected one of: full,
ddl, readonly.") from exc
+ raise ValueError(
+ "Unsupported policy preset. Expected one of: full, ddl, readonly."
+ ) from exc
def supported_policy_keys() -> list[str]:
- return sorted(_POLICY_KEYS)
+ return sorted(_POLICY_KEYS - _ADMIN_POLICY_KEYS)
+
+
+def _no_broader(key: str, candidate: str, limit: str) -> bool:
+ if candidate == limit:
+ return True
+ if key.startswith("IOTDB_ENABLE_"):
+ return candidate == "false"
+ if "CONFIRM" in key:
+ return candidate == "true"
+ if key == "IOTDB_SQL_DRIVER_MODE":
+ return _MODE_RANK[candidate] <= _MODE_RANK[limit]
+ if key.endswith("_ALLOWED_USERS"):
+ candidate_users = {item for item in candidate.split(",") if item}
+ limit_users = {item for item in limit.split(",") if item}
+ return "*" in limit_users or candidate_users <= limit_users
+ return False
+
+
+def _apply_session_candidate(candidate: dict[str, str]) -> dict[str, Any]:
+ """Validate, authorize, and commit under one lock, including
reset/replace."""
+ global _POLICY_REVISION
+ assert _DEPLOYMENT_POLICY is not None
+ before = {**_DEPLOYMENT_POLICY, **_SESSION_POLICY}
+ after = {**_DEPLOYMENT_POLICY, **candidate}
+ above_ceiling = [
+ key
+ for key in after
+ if not _no_broader(key, after[key], _DEPLOYMENT_POLICY[key])
+ ]
+ if above_ceiling:
+ raise PermissionError(
+ "Deployment permission ceiling exceeded: "
+ + ", ".join(sorted(above_ceiling))
+ + ". Administrator must change protected configuration and
restart."
+ )
+ changes = {
+ key: {"before": before[key], "after": after[key]}
+ for key in sorted(after)
+ if before[key] != after[key]
+ }
+ widened = [key for key in changes if not _no_broader(key, after[key],
before[key])]
+ if widened and _APPROVAL_MODE == "require":
+ if _APPROVAL_STORE is None:
+ return {
+ **dynamic_policy_snapshot(),
+ "status": "approval_unavailable",
+ "applied": False,
+ "requested_changes": changes,
+ "message": "No administrator approval directory configured;
policy unchanged.",
+ }
+ fingerprint = hashlib.sha256(
+ json.dumps(_DEPLOYMENT_POLICY, sort_keys=True).encode()
+ ).hexdigest()
+ approval = _APPROVAL_STORE.authorize(
+ {
+ "revision": _POLICY_REVISION,
+ "deployment_fingerprint": fingerprint,
+ "changes": changes,
+ "widened_keys": widened,
+ }
+ )
+ if approval["status"] != "approved":
+ return {
+ **dynamic_policy_snapshot(),
+ **approval,
+ "applied": False,
+ "message": "Policy unchanged. Administrator approval is
required before retrying.",
+ }
+ _SESSION_POLICY.clear()
+ _SESSION_POLICY.update(candidate)
+ if changes:
+ _POLICY_REVISION += 1
+ return {**dynamic_policy_snapshot(), "status": "applied", "applied": True}
def set_session_policy(
@@ -295,59 +464,56 @@ def set_session_policy(
preset: str | None = None,
replace: bool = False,
) -> dict[str, Any]:
- updates: dict[str, str] = {}
- if preset:
- updates.update(session_policy_for_preset(preset))
- if policy:
- _validate_policy_keys(policy)
- updates.update({str(key): _coerce_policy_value(value) for key, value
in policy.items()})
-
+ initialize_runtime_policy()
with _SESSION_POLICY_LOCK:
- preserved_enforcement = (
- {
- key: value
- for key, value in _SESSION_POLICY.items()
- if key in _ENFORCEMENT_POLICY_KEYS
- }
- if replace and preset
- else {}
- )
- if replace:
- _SESSION_POLICY.clear()
- _SESSION_POLICY.update(preserved_enforcement)
- _SESSION_POLICY.update(updates)
-
- return dynamic_policy_snapshot()
+ assert _DEPLOYMENT_POLICY is not None
+ updates: dict[str, str] = {}
+ if preset:
+ # Presets are permission caps, intersected with operator
restrictions.
+ for key, value in session_policy_for_preset(preset).items():
+ ceiling = _DEPLOYMENT_POLICY[key]
+ updates[key] = value if _no_broader(key, value, ceiling) else
ceiling
+ if policy:
+ _validate_policy_keys(policy)
+ updates.update(
+ {
+ str(key): _normalize_policy_value(str(key), value)
+ for key, value in policy.items()
+ }
+ )
+ candidate = {} if replace else dict(_SESSION_POLICY)
+ candidate.update(updates)
+ return _apply_session_candidate(candidate)
def reset_session_policy(keys: list[str] | None = None) -> dict[str, Any]:
+ initialize_runtime_policy()
if keys is not None:
_validate_policy_keys({key: "" for key in keys})
with _SESSION_POLICY_LOCK:
- if keys is None:
- _SESSION_POLICY.clear()
- else:
+ candidate = {} if keys is None else dict(_SESSION_POLICY)
+ if keys is not None:
for key in keys:
- _SESSION_POLICY.pop(key, None)
-
- return dynamic_policy_snapshot()
+ candidate.pop(key, None)
+ return _apply_session_candidate(candidate)
def dynamic_policy_snapshot() -> dict[str, Any]:
- env, path = _load_mcp_env()
- session_policy = _session_policy_copy()
- effective: dict[str, str] = {}
- for key in supported_policy_keys():
- value = dynamic_getenv(key, "")
- if value not in (None, ""):
- effective[key] = value
- return {
- "mcp_config_path": path,
- "session_policy": _redact_env(session_policy),
- "mcp_env": _redact_env(env),
- "defaults": {**_FULL_PERMISSION_DEFAULTS, **_ENFORCEMENT_DEFAULTS},
- "effective_policy": _redact_env(effective),
- "supported_keys": supported_policy_keys(),
- "presets": sorted(_PRESET_POLICIES),
- }
+ initialize_runtime_policy()
+ with _SESSION_POLICY_LOCK:
+ assert _DEPLOYMENT_POLICY is not None
+ return {
+ "mcp_config_path": _DEPLOYMENT_CONFIG_PATH,
+ "session_policy": dict(_SESSION_POLICY),
+ "mcp_env": _redact_env(_DEPLOYMENT_ENV),
+ "defaults": {**_FULL_PERMISSION_DEFAULTS, **_ENFORCEMENT_DEFAULTS},
+ "deployment_policy": dict(_DEPLOYMENT_POLICY),
+ "effective_policy": {**_DEPLOYMENT_POLICY, **_SESSION_POLICY},
+ "supported_keys": supported_policy_keys(),
+ "administrator_only_keys": sorted(_ADMIN_POLICY_KEYS),
+ "approval_mode": _APPROVAL_MODE,
+ "approval_available": _APPROVAL_STORE is not None,
+ "policy_revision": _POLICY_REVISION,
+ "presets": sorted(_PRESET_POLICIES),
+ }
diff --git a/src/iotdb_mcp_server/server.py b/src/iotdb_mcp_server/server.py
index b0935e9..f86dd85 100644
--- a/src/iotdb_mcp_server/server.py
+++ b/src/iotdb_mcp_server/server.py
@@ -21,6 +21,7 @@ import logging
from fastmcp import FastMCP
from iotdb_mcp_server.config import Config
+from iotdb_mcp_server.runtime_policy import initialize_runtime_policy
from iotdb_mcp_server.services.database import register_database_tools
from iotdb_mcp_server.services.explain import register_explain_tools
from iotdb_mcp_server.services.metadata import register_metadata_tools
@@ -44,6 +45,7 @@ logging.basicConfig(
logger = logging.getLogger("iotdb_mcp_server")
config = Config.from_env_arguments()
+initialize_runtime_policy()
logger.info(
"IoTDB Config: %s",
diff --git a/src/iotdb_mcp_server/services/README.md
b/src/iotdb_mcp_server/services/README.md
index 27d1dad..ce0d2e7 100644
--- a/src/iotdb_mcp_server/services/README.md
+++ b/src/iotdb_mcp_server/services/README.md
@@ -107,11 +107,15 @@ Features:
- `ddl`: advisory signal for read and DDL-capable workflows
- `readonly`: advisory signal for read-only workflows
- In `strict` mode only, these presets become hard execution gates
-- Effective policy priority:
- - session overlay set by MCP tool
- - dynamically read `.mcp.json`
- - process environment
- - built-in full-permission defaults
+- Deployment ceiling is frozen at server startup: process environment, then
+ `.mcp.json`, then defaults. Policy files are not hot-reloaded.
+- Every session value must be no broader than that ceiling. Presets are
intersected
+ with it; explicit above-ceiling requests are rejected atomically.
+- Narrowing applies immediately. Widening (including reset/replace) requires
+ out-of-band approval by default. The operator may explicitly configure
+ `IOTDB_SESSION_POLICY_APPROVAL_MODE=allow` for in-ceiling changes only.
+- Enforcement switches and SQL classification extensions are
administrator-only.
+- See [session policy
administration](../../../docs/session-policy-security.md).
### `metadata.py`
@@ -284,18 +288,18 @@ Parser check:
## Security and Permission Model
-IoTDB MCP permissions default to host-agent approval flow. The MCP server
reports
-required permission and risk; Codex, Claude Code, or OpenCode owns user
approval
-through its normal question/approval mechanism. This keeps MCP policy aligned
-with the host agent system instead of turning `.mcp.json` into a hard blocker.
+SQL execution permissions remain advisory by default. The MCP server reports
+required permission and risk, but a tool-supplied confirmation boolean does not
+prove human approval. For a service-enforced SQL permission boundary, enable
strict
+mode. Session policy mutation authorization is enforced in both modes.
Default mode is `advisory`. In advisory mode:
- `IOTDB_SQL_DRIVER_MODE` is a prompt-layer signal, not an execution blocker.
- `IOTDB_ENABLE_*` and `*_ALLOWED_USERS` are prompt-layer/session settings, not
hard gates.
-- Destructive operations still require explicit tool confirmation flags, such
- as `confirm_destructive=true`, after the host agent has obtained approval.
+- Destructive operations still require tool confirmation flags, such as
+ `confirm_destructive=true`. These are caller assertions, not verified
approvals.
Strict hard gating is available only when explicitly enabled:
@@ -305,7 +309,7 @@ TIMESEEK_MCP_PERMISSION_ENFORCEMENT=strict
IOTDB_STRICT_PERMISSION_ENFORCEMENT=true
```
-Policy can be changed at runtime without restarting the MCP server:
+Session policy can be narrowed at runtime without restarting the MCP server:
```json
{
@@ -314,7 +318,7 @@ Policy can be changed at runtime without restarting the MCP
server:
}
```
-or:
+Widening within the startup deployment ceiling can be requested with:
```json
{
@@ -325,6 +329,14 @@ or:
}
```
+The second example may return `applied=false` with `status=approval_required`,
+`approval_rejected`, or `approval_unavailable`. It must not be treated as an
applied
+change. An administrator reviews and decides the exact request through
+`iotdb-policy-admin`, outside the agent's authority, then the client retries.
+Changing the deployment ceiling, either enforcement switch, or SQL
classification
+prefixes requires administrator-controlled configuration and a server restart.
+See [the complete workflow](../../../docs/session-policy-security.md).
+
### Per-module advisory keys
- `database.py`
diff --git a/src/iotdb_mcp_server/services/database.py
b/src/iotdb_mcp_server/services/database.py
index 5578c1e..4297dbe 100644
--- a/src/iotdb_mcp_server/services/database.py
+++ b/src/iotdb_mcp_server/services/database.py
@@ -64,7 +64,7 @@ def _assert_database_ddl_permission(
"Set IOTDB_ENABLE_DATABASE_DDL=true to enable."
)
- allowed_users =
_csv_set(dynamic_getenv("IOTDB_DATABASE_DDL_ALLOWED_USERS", "root") or "root")
+ allowed_users =
_csv_set(dynamic_getenv("IOTDB_DATABASE_DDL_ALLOWED_USERS", "root") or "")
if "*" not in allowed_users and config.user not in allowed_users:
raise PermissionError(
f"Current MCP user '{config.user}' is not allowed by
IOTDB_DATABASE_DDL_ALLOWED_USERS."
diff --git a/src/iotdb_mcp_server/services/metadata.py
b/src/iotdb_mcp_server/services/metadata.py
index b7a8398..7341bd4 100644
--- a/src/iotdb_mcp_server/services/metadata.py
+++ b/src/iotdb_mcp_server/services/metadata.py
@@ -59,7 +59,7 @@ def _assert_metadata_permission(config: Config) -> None:
)
allowed_users = _csv_set(
- dynamic_getenv("IOTDB_METADATA_ALLOWED_USERS", "*") or "*"
+ dynamic_getenv("IOTDB_METADATA_ALLOWED_USERS", "*") or ""
)
if "*" not in allowed_users and config.user not in allowed_users:
raise PermissionError(
diff --git a/src/iotdb_mcp_server/services/model.py
b/src/iotdb_mcp_server/services/model.py
index 577c172..48a38dc 100644
--- a/src/iotdb_mcp_server/services/model.py
+++ b/src/iotdb_mcp_server/services/model.py
@@ -69,7 +69,7 @@ def _assert_model_permission(
)
allowed_users = _csv_set(
- dynamic_getenv("IOTDB_MODEL_ALLOWED_USERS", "root") or "root"
+ dynamic_getenv("IOTDB_MODEL_ALLOWED_USERS", "root") or ""
)
if "*" not in allowed_users and config.user not in allowed_users:
raise PermissionError(
diff --git a/src/iotdb_mcp_server/services/runtime_policy.py
b/src/iotdb_mcp_server/services/runtime_policy.py
index c768389..32015be 100644
--- a/src/iotdb_mcp_server/services/runtime_policy.py
+++ b/src/iotdb_mcp_server/services/runtime_policy.py
@@ -48,13 +48,18 @@ def register_runtime_policy_tools(mcp, config: Config,
logger: logging.Logger) -
replace: bool = False,
) -> list[TextContent]:
"""
- Set session-scoped IoTDB MCP permissions without restarting the MCP
server.
+ Request session permissions bounded by the immutable deployment policy.
- preset accepts: full, ddl, readonly. Explicit policy values override
preset values.
+ Narrowing applies immediately. Widening may return approval_required
with
+ no changes applied; an administrator must approve out of band before
retry.
+ There is no agent-supplied approval flag. Presets (full, ddl,
readonly) are
+ intersected with the deployment ceiling. Explicit values cannot exceed
it.
+ Enforcement switches and SQL classification prefixes are
administrator-only.
"""
snapshot = set_session_policy(policy=policy, preset=preset,
replace=replace)
logger.info(
- "Updated IoTDB session policy preset=%s replace=%s keys=%s",
+ "IoTDB session policy status=%s preset=%s replace=%s keys=%s",
+ snapshot["status"],
preset,
replace,
sorted((policy or {}).keys()),
@@ -62,16 +67,22 @@ def register_runtime_policy_tools(mcp, config: Config,
logger: logging.Logger) -
return payload_response(
"set_iotdb_session_policy",
snapshot,
- message="IoTDB session policy updated.",
+ message="IoTDB session policy: " + snapshot["status"] + ".",
)
@mcp.tool()
- async def reset_iotdb_session_policy(keys: list[str] | None = None) ->
list[TextContent]:
- """Clear all or selected session-scoped IoTDB MCP permission
overrides."""
+ async def reset_iotdb_session_policy(
+ keys: list[str] | None = None,
+ ) -> list[TextContent]:
+ """Reset overrides; any resulting widening requires the same approval
as set."""
snapshot = reset_session_policy(keys=keys)
- logger.info("Reset IoTDB session policy keys=%s", keys or "*")
+ logger.info(
+ "IoTDB session policy reset status=%s keys=%s",
+ snapshot["status"],
+ keys or "*",
+ )
return payload_response(
"reset_iotdb_session_policy",
snapshot,
- message="IoTDB session policy reset.",
+ message="IoTDB session policy reset: " + snapshot["status"] + ".",
)
diff --git a/src/iotdb_mcp_server/services/sql_driver.py
b/src/iotdb_mcp_server/services/sql_driver.py
index 055c3e5..5636d52 100644
--- a/src/iotdb_mcp_server/services/sql_driver.py
+++ b/src/iotdb_mcp_server/services/sql_driver.py
@@ -527,7 +527,8 @@ def _sql_permission_payload(
"Call sql_execute with confirm_destructive=true when the "
"confirmation_parameter is confirm_destructive."
),
- "do_not_use_environment_as_primary_gate": True,
+ "deployment_policy_is_immutable": True,
+ "strict_policy_cannot_be_bypassed_by_confirmation": True,
},
}
@@ -547,7 +548,7 @@ def _assert_sql_driver_permission(
)
allowed_users = _csv_set(
- dynamic_getenv("IOTDB_SQL_DRIVER_ALLOWED_USERS", "root") or "root"
+ dynamic_getenv("IOTDB_SQL_DRIVER_ALLOWED_USERS", "root") or ""
)
if "*" not in allowed_users and config.user not in allowed_users:
raise PermissionError(
diff --git a/src/iotdb_mcp_server/services/table.py
b/src/iotdb_mcp_server/services/table.py
index 958a05f..2eb8fd2 100644
--- a/src/iotdb_mcp_server/services/table.py
+++ b/src/iotdb_mcp_server/services/table.py
@@ -50,7 +50,7 @@ def _assert_table_ddl_permission(
"Set IOTDB_ENABLE_TABLE_DDL=true to enable."
)
- allowed_users =
_csv_set(dynamic_getenv("IOTDB_TABLE_DDL_ALLOWED_USERS", "root") or "root")
+ allowed_users =
_csv_set(dynamic_getenv("IOTDB_TABLE_DDL_ALLOWED_USERS", "root") or "")
if "*" not in allowed_users and config.user not in allowed_users:
raise PermissionError(
f"Current MCP user '{config.user}' is not allowed by
IOTDB_TABLE_DDL_ALLOWED_USERS."
diff --git a/src/iotdb_mcp_server/services/timeseries.py
b/src/iotdb_mcp_server/services/timeseries.py
index 89b034e..6678af1 100644
--- a/src/iotdb_mcp_server/services/timeseries.py
+++ b/src/iotdb_mcp_server/services/timeseries.py
@@ -52,7 +52,7 @@ def _assert_timeseries_ddl_permission(
"Set IOTDB_ENABLE_TIMESERIES_DDL=true to enable."
)
- allowed_users =
_csv_set(dynamic_getenv("IOTDB_TIMESERIES_DDL_ALLOWED_USERS", "root") or "root")
+ allowed_users =
_csv_set(dynamic_getenv("IOTDB_TIMESERIES_DDL_ALLOWED_USERS", "root") or "")
if "*" not in allowed_users and config.user not in allowed_users:
raise PermissionError(
f"Current MCP user '{config.user}' is not allowed by
IOTDB_TIMESERIES_DDL_ALLOWED_USERS."
diff --git a/src/iotdb_mcp_server/services/ttl.py
b/src/iotdb_mcp_server/services/ttl.py
index b22786e..43fea71 100644
--- a/src/iotdb_mcp_server/services/ttl.py
+++ b/src/iotdb_mcp_server/services/ttl.py
@@ -55,7 +55,7 @@ def _assert_ttl_permission(
"Set IOTDB_ENABLE_TTL_SQL=true to enable."
)
- allowed_users = _csv_set(dynamic_getenv("IOTDB_TTL_ALLOWED_USERS",
"root") or "root")
+ allowed_users = _csv_set(dynamic_getenv("IOTDB_TTL_ALLOWED_USERS",
"root") or "")
if "*" not in allowed_users and config.user not in allowed_users:
raise PermissionError(
f"Current MCP user '{config.user}' is not allowed by
IOTDB_TTL_ALLOWED_USERS."
diff --git a/src/iotdb_mcp_server/services/write.py
b/src/iotdb_mcp_server/services/write.py
index 5250112..4396473 100644
--- a/src/iotdb_mcp_server/services/write.py
+++ b/src/iotdb_mcp_server/services/write.py
@@ -52,7 +52,7 @@ def _assert_write_permission(
"Set IOTDB_ENABLE_WRITE_DML=true to enable."
)
- allowed_users = _csv_set(dynamic_getenv("IOTDB_WRITE_ALLOWED_USERS",
"root") or "root")
+ allowed_users = _csv_set(dynamic_getenv("IOTDB_WRITE_ALLOWED_USERS",
"root") or "")
if "*" not in allowed_users and config.user not in allowed_users:
raise PermissionError(
f"Current MCP user '{config.user}' is not allowed by
IOTDB_WRITE_ALLOWED_USERS."
diff --git a/tests/test_runtime_policy.py b/tests/test_runtime_policy.py
index df1e213..0ea7676 100644
--- a/tests/test_runtime_policy.py
+++ b/tests/test_runtime_policy.py
@@ -1,8 +1,16 @@
from __future__ import annotations
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+import contextlib
+import io
+import json
+import logging
import os
+import subprocess
import sys
import unittest
+import tempfile
from pathlib import Path
from unittest.mock import patch
@@ -14,12 +22,61 @@ from iotdb_mcp_server.runtime_policy import ( # noqa: E402
reset_session_policy,
set_session_policy,
strict_permission_enforcement,
+ dynamic_getenv,
+ initialize_runtime_policy,
)
+from iotdb_mcp_server import runtime_policy as runtime # noqa: E402
+from iotdb_mcp_server.policy_approval import main as administer # noqa: E402
+from iotdb_mcp_server.config import Config # noqa: E402
+from iotdb_mcp_server.services.sql_driver import (
+ _assert_sql_driver_permission,
+) # noqa: E402
+from iotdb_mcp_server.services.write import _assert_write_permission # noqa:
E402
+from iotdb_mcp_server.services.runtime_policy import (
+ register_runtime_policy_tools,
+) # noqa: E402
class RuntimePolicyTest(unittest.TestCase):
- def tearDown(self) -> None:
- reset_session_policy()
+ def setUp(self) -> None:
+ self.enterContext(patch.dict(os.environ, {}, clear=True))
+ self.enterContext(patch.object(runtime, "_load_mcp_env",
return_value=({}, "")))
+ for name, value in {
+ "_DEPLOYMENT_POLICY": None,
+ "_DEPLOYMENT_ENV": {},
+ "_DEPLOYMENT_CONFIG_PATH": "",
+ "_SESSION_POLICY": {},
+ "_POLICY_REVISION": 0,
+ "_APPROVAL_STORE": None,
+ "_APPROVAL_MODE": "require",
+ }.items():
+ self.enterContext(patch.object(runtime, name, value))
+
+ def approval_directory(self) -> Path:
+ directory = Path(self.enterContext(tempfile.TemporaryDirectory()))
+ os.environ["IOTDB_SESSION_POLICY_APPROVAL_DIR"] = str(directory)
+ return directory
+
+ def decide(self, directory: Path, request: dict, action: str = "approve")
-> None:
+ with contextlib.redirect_stdout(io.StringIO()):
+ self.assertEqual(
+ administer(
+ [action, request["request_id"], "--directory",
str(directory)]
+ ),
+ 0,
+ )
+
+ def config(self) -> Config:
+ return Config(
+ host="127.0.0.1",
+ port=6667,
+ user="root",
+ password="test",
+ database="test",
+ sql_dialect="tree",
+ timezone="UTC",
+ export_path="/tmp",
+ )
def test_targets_json_is_fully_withheld_from_policy_snapshot(self) -> None:
mcp_env = {
@@ -61,22 +118,349 @@ class RuntimePolicyTest(unittest.TestCase):
snapshot["session_policy"],
)
- def test_replacing_with_preset_preserves_session_enforcement(self) -> None:
- with (
- patch(
- "iotdb_mcp_server.runtime_policy._load_mcp_env",
- return_value=({}, ""),
- ),
- patch.dict(os.environ, {}, clear=True),
+ def test_both_enforcement_switches_and_classification_are_admin_only(self)
-> None:
+ os.environ["IOTDB_STRICT_PERMISSION_ENFORCEMENT"] = "true"
+ for key, value in {
+ "IOTDB_STRICT_PERMISSION_ENFORCEMENT": False,
+ "TIMESEEK_MCP_PERMISSION_ENFORCEMENT": "advisory",
+ "IOTDB_SQL_DRIVER_EXTRA_READONLY_PREFIXES": "INSERT INTO",
+ "IOTDB_SQL_DRIVER_EXTRA_DDL_PREFIXES": "DELETE FROM",
+ "IOTDB_SQL_DRIVER_EXTRA_FULL_PREFIXES": "GRANT",
+ }.items():
+ with self.subTest(key=key):
+ with self.assertRaises(PermissionError):
+ set_session_policy({key: value}, replace=True)
+ with self.assertRaises(PermissionError):
+ reset_session_policy([key])
+ self.assertTrue(strict_permission_enforcement())
+ self.assertEqual(dynamic_policy_snapshot()["session_policy"], {})
+
+ def test_environment_ceiling_is_frozen_and_takes_precedence_over_config(
+ self,
+ ) -> None:
+ os.environ.update(
+ {
+ "IOTDB_STRICT_PERMISSION_ENFORCEMENT": "true",
+ "IOTDB_SQL_DRIVER_MODE": "readonly",
+ "IOTDB_ENABLE_WRITE_DML": "false",
+ }
+ )
+ runtime._load_mcp_env.return_value = (
+ {
+ "IOTDB_SQL_DRIVER_MODE": "full",
+ "IOTDB_ENABLE_WRITE_DML": "true",
+ },
+ "/tmp/.mcp.json",
+ )
+ initialize_runtime_policy()
+ os.environ["IOTDB_SQL_DRIVER_MODE"] = "full"
+ os.environ["IOTDB_STRICT_PERMISSION_ENFORCEMENT"] = "false"
+ runtime._load_mcp_env.return_value = ({"IOTDB_SQL_DRIVER_MODE":
"full"}, "")
+ with self.assertRaisesRegex(PermissionError, "ceiling"):
+ set_session_policy({"IOTDB_SQL_DRIVER_MODE": "full"})
+ snapshot = set_session_policy(preset="full")
+ self.assertEqual(
+ snapshot["effective_policy"]["IOTDB_SQL_DRIVER_MODE"], "readonly"
+ )
+ self.assertEqual(dynamic_getenv("IOTDB_ENABLE_WRITE_DML"), "false")
+ self.assertTrue(strict_permission_enforcement())
+ with self.assertRaises(PermissionError):
+ _assert_sql_driver_permission(
+ self.config(), "ddl", "readonly", True, "DROP DATABASE"
+ )
+ with self.assertRaises(PermissionError):
+ _assert_write_permission(self.config(), "INSERT")
+
+ def test_preset_intersects_allowlists_and_confirmation_ceiling(self) ->
None:
+ os.environ["IOTDB_WRITE_ALLOWED_USERS"] = "alice"
+ os.environ["IOTDB_REQUIRE_DELETE_CONFIRM"] = "true"
+ result = set_session_policy(preset="readonly")
+ self.assertTrue(result["applied"])
+ self.assertEqual(dynamic_getenv("IOTDB_WRITE_ALLOWED_USERS"), "alice")
+ self.assertEqual(dynamic_getenv("IOTDB_ENABLE_MODEL_MANAGEMENT"),
"false")
+ for policy in (
+ {"IOTDB_WRITE_ALLOWED_USERS": "*"},
+ {"IOTDB_REQUIRE_DELETE_CONFIRM": False},
):
- set_session_policy({"TIMESEEK_MCP_PERMISSION_ENFORCEMENT":
"strict"})
- snapshot = set_session_policy(preset="readonly", replace=True)
+ with self.assertRaises(PermissionError):
+ set_session_policy(policy)
- self.assertTrue(strict_permission_enforcement())
- self.assertEqual(
-
snapshot["session_policy"]["TIMESEEK_MCP_PERMISSION_ENFORCEMENT"],
- "strict",
+ def test_narrowing_applies_and_widening_fails_closed_without_channel(self)
-> None:
+ result = set_session_policy({"IOTDB_SQL_DRIVER_MODE": "readonly"})
+ self.assertTrue(result["applied"])
+ for operation in (
+ lambda: set_session_policy({"IOTDB_SQL_DRIVER_MODE": "full"}),
+ lambda: set_session_policy(preset="full", replace=True),
+ lambda: set_session_policy(replace=True),
+ lambda: reset_session_policy(),
+ lambda: reset_session_policy(["IOTDB_SQL_DRIVER_MODE"]),
+ ):
+ with self.subTest(operation=operation):
+ result = operation()
+ self.assertFalse(result["applied"])
+ self.assertEqual(result["status"], "approval_unavailable")
+ self.assertEqual(dynamic_getenv("IOTDB_SQL_DRIVER_MODE"),
"readonly")
+
+ def test_approval_is_exact_one_time_and_precedes_mutation(self) -> None:
+ directory = self.approval_directory()
+ set_session_policy(preset="readonly")
+ result = set_session_policy(preset="full")
+ self.assertEqual(result["status"], "approval_required")
+ self.assertFalse(result["applied"])
+ self.assertEqual(dynamic_getenv("IOTDB_ENABLE_WRITE_DML"), "false")
+ self.decide(directory, result)
+ approved = set_session_policy(preset="full")
+ self.assertTrue(approved["applied"])
+ self.assertEqual(dynamic_getenv("IOTDB_ENABLE_WRITE_DML"), "true")
+ self.assertEqual(list(directory.iterdir()), [])
+ set_session_policy(preset="readonly")
+ replay = set_session_policy(preset="full")
+ self.assertFalse(replay["applied"])
+ self.assertNotEqual(result["request_id"], replay["request_id"])
+
+ def test_reset_and_replace_require_approval_before_restoring_permissions(
+ self,
+ ) -> None:
+ directory = self.approval_directory()
+ for operation in (
+ reset_session_policy,
+ lambda: set_session_policy(replace=True),
+ ):
+ with self.subTest(operation=operation):
+ set_session_policy({"IOTDB_ENABLE_WRITE_DML": False})
+ result = operation()
+ self.assertFalse(result["applied"])
+ self.decide(directory, result)
+ self.assertTrue(operation()["applied"])
+ self.assertEqual(dynamic_getenv("IOTDB_ENABLE_WRITE_DML"),
"true")
+
+ def test_allow_is_operator_only_and_never_bypasses_ceiling(self) -> None:
+ os.environ.update(
+ {
+ "IOTDB_SESSION_POLICY_APPROVAL_MODE": "allow",
+ "IOTDB_ENABLE_DATABASE_DDL": "false",
+ }
+ )
+ set_session_policy(preset="readonly")
+ self.assertTrue(set_session_policy(preset="full")["applied"])
+ with self.assertRaises(PermissionError):
+ set_session_policy({"IOTDB_ENABLE_DATABASE_DDL": True})
+ for key in (
+ "IOTDB_SESSION_POLICY_APPROVAL_MODE",
+ "IOTDB_SESSION_POLICY_APPROVAL_DIR",
+ "approved",
+ ):
+ with self.assertRaises(ValueError):
+ set_session_policy({key: "allow"})
+
+ def test_approval_mode_is_frozen(self) -> None:
+ set_session_policy(preset="readonly")
+ os.environ["IOTDB_SESSION_POLICY_APPROVAL_MODE"] = "allow"
+ self.assertFalse(reset_session_policy()["applied"])
+
+ def test_rejection_keeps_original_policy(self) -> None:
+ directory = self.approval_directory()
+ set_session_policy(preset="readonly")
+ request = reset_session_policy()
+ self.decide(directory, request, "deny")
+ result = reset_session_policy()
+ self.assertEqual(result["status"], "approval_rejected")
+ self.assertFalse(result["applied"])
+ self.assertEqual(dynamic_getenv("IOTDB_SQL_DRIVER_MODE"), "readonly")
+
+ def test_expired_approval_cannot_apply(self) -> None:
+ directory = self.approval_directory()
+ set_session_policy(preset="readonly")
+ request = reset_session_policy()
+ self.decide(directory, request)
+ with patch(
+ "iotdb_mcp_server.policy_approval.time.time",
+ return_value=request["expires_at"] + 1,
+ ):
+ result = reset_session_policy()
+ self.assertFalse(result["applied"])
+ self.assertNotEqual(result["request_id"], request["request_id"])
+
+ def test_changed_revision_invalidates_pending_approval(self) -> None:
+ directory = self.approval_directory()
+ set_session_policy({"IOTDB_ENABLE_WRITE_DML": False})
+ request = reset_session_policy()
+ self.decide(directory, request)
+ set_session_policy({"IOTDB_ENABLE_DATABASE_DDL": False})
+ result = reset_session_policy()
+ self.assertFalse(result["applied"])
+ self.assertNotEqual(result["request_id"], request["request_id"])
+
+ def test_different_proposal_cannot_reuse_approval(self) -> None:
+ directory = self.approval_directory()
+ set_session_policy(preset="readonly")
+ request = set_session_policy({"IOTDB_ENABLE_WRITE_DML": True})
+ self.decide(directory, request)
+ result = set_session_policy(preset="full")
+ self.assertFalse(result["applied"])
+ self.assertNotEqual(result["request_id"], request["request_id"])
+
+ def test_tampered_request_cannot_approve_original_proposal(self) -> None:
+ directory = self.approval_directory()
+ set_session_policy(preset="readonly")
+ request = reset_session_policy()
+ path = directory / f"{request['request_id']}.request.json"
+ data = json.loads(path.read_text())
+ data["changes"] = {}
+ path.write_text(json.dumps(data))
+ self.decide(directory, request)
+ with self.assertRaises(PermissionError):
+ reset_session_policy()
+ self.assertEqual(dynamic_getenv("IOTDB_SQL_DRIVER_MODE"), "readonly")
+
+ def test_atomic_rejection_and_invalid_values(self) -> None:
+ os.environ["IOTDB_ENABLE_DATABASE_DDL"] = "false"
+ with self.assertRaises(PermissionError):
+ set_session_policy(
+ {"IOTDB_ENABLE_WRITE_DML": False, "IOTDB_ENABLE_DATABASE_DDL":
True}
)
+ self.assertEqual(dynamic_policy_snapshot()["session_policy"], {})
+ for policy in (
+ {"IOTDB_ENABLE_WRITE_DML": "maybe"},
+ {"IOTDB_SQL_DRIVER_MODE": "unrestricted"},
+ ):
+ with self.assertRaises(ValueError):
+ set_session_policy(policy)
+ self.assertEqual(dynamic_policy_snapshot()["session_policy"], {})
+
+ def test_empty_allowlist_denies_root_and_restoring_it_requires_approval(
+ self,
+ ) -> None:
+ os.environ["IOTDB_STRICT_PERMISSION_ENFORCEMENT"] = "true"
+ set_session_policy({"IOTDB_WRITE_ALLOWED_USERS": ""})
+ self.assertEqual(dynamic_getenv("IOTDB_WRITE_ALLOWED_USERS", "root"),
"")
+ with self.assertRaises(PermissionError):
+ _assert_write_permission(self.config(), "INSERT")
+ self.assertFalse(reset_session_policy()["applied"])
+
+ def test_confirmation_and_allowlist_widening_need_approval(self) -> None:
+ os.environ["IOTDB_REQUIRE_DELETE_CONFIRM"] = "false"
+ for narrow, wider in (
+ (
+ {"IOTDB_REQUIRE_DELETE_CONFIRM": True},
+ {"IOTDB_REQUIRE_DELETE_CONFIRM": False},
+ ),
+ (
+ {"IOTDB_WRITE_ALLOWED_USERS": "alice"},
+ {"IOTDB_WRITE_ALLOWED_USERS": "alice,bob"},
+ ),
+ ({"IOTDB_SQL_DRIVER_MODE": "readonly"}, {"IOTDB_SQL_DRIVER_MODE":
"ddl"}),
+ ):
+ self.assertTrue(set_session_policy(narrow)["applied"])
+ self.assertFalse(set_session_policy(wider)["applied"])
+
+ def test_invalid_deployment_fails_initialization(self) -> None:
+ os.environ["IOTDB_ENABLE_WRITE_DML"] = "maybe"
+ with self.assertRaises(ValueError):
+ initialize_runtime_policy()
+ self.assertIsNone(runtime._DEPLOYMENT_POLICY)
+
+ def test_restart_invalidates_approval(self) -> None:
+ directory = self.approval_directory()
+ set_session_policy(preset="readonly")
+ request = reset_session_policy()
+ self.decide(directory, request)
+ runtime._APPROVAL_STORE = runtime.PolicyApprovalStore(directory)
+ result = reset_session_policy()
+ self.assertFalse(result["applied"])
+ self.assertNotEqual(result["request_id"], request["request_id"])
+
+ def test_concurrent_retries_consume_approval_once(self) -> None:
+ directory = self.approval_directory()
+ set_session_policy(preset="readonly")
+ request = reset_session_policy()
+ self.decide(directory, request)
+ with ThreadPoolExecutor(max_workers=2) as executor:
+ results = list(executor.map(lambda _: reset_session_policy(),
range(2)))
+ self.assertTrue(all(result["applied"] for result in results))
+ self.assertEqual(dynamic_policy_snapshot()["policy_revision"], 2)
+ self.assertEqual(list(directory.iterdir()), [])
+
+ def test_pending_approval_count_is_bounded(self) -> None:
+ self.approval_directory()
+ set_session_policy({"IOTDB_WRITE_ALLOWED_USERS": "alice"})
+ with patch("iotdb_mcp_server.policy_approval._MAX_PENDING", 2):
+ for user in ("bob", "carol"):
+ self.assertFalse(
+ set_session_policy({"IOTDB_WRITE_ALLOWED_USERS":
f"alice,{user}"})[
+ "applied"
+ ]
+ )
+ with self.assertRaisesRegex(PermissionError, "Too many"):
+ set_session_policy({"IOTDB_WRITE_ALLOWED_USERS": "*"})
+
+ def test_approval_directory_and_cli_request_id_are_validated(self) -> None:
+ directory = self.approval_directory()
+ if os.name == "posix":
+ directory.chmod(0o777)
+ try:
+ with self.assertRaises(PermissionError):
+ initialize_runtime_policy()
+ finally:
+ directory.chmod(0o700)
+ with contextlib.redirect_stderr(io.StringIO()),
self.assertRaises(SystemExit):
+ administer(["approve", "../invalid", "--directory",
str(directory)])
+
+ def test_real_mcp_tool_approval_flow(self) -> None:
+ from fastmcp import Client, FastMCP
+
+ directory = self.approval_directory()
+ mcp = FastMCP("policy-test")
+ register_runtime_policy_tools(mcp, self.config(),
logging.getLogger(__name__))
+
+ async def exercise():
+ async with Client(mcp) as client:
+ tools = await client.list_tools()
+ self.assertEqual(
+ {tool.name for tool in tools},
+ {
+ "get_iotdb_session_policy",
+ "set_iotdb_session_policy",
+ "reset_iotdb_session_policy",
+ },
+ )
+ setter = next(
+ tool for tool in tools if tool.name ==
"set_iotdb_session_policy"
+ )
+ self.assertEqual(
+ set(setter.inputSchema["properties"]),
+ {"policy", "preset", "replace"},
+ )
+ await client.call_tool(
+ "set_iotdb_session_policy", {"preset": "readonly"}
+ )
+ response = await
client.call_tool("reset_iotdb_session_policy", {})
+ pending = json.loads(response.content[0].text)["payload"]
+ self.assertFalse(pending["applied"])
+ # Exercise the independent administrative process, not a tool
or
+ # an in-process callback that an agent could supply itself.
+ decision = subprocess.run(
+ [
+ sys.executable,
+ "-m",
+ "iotdb_mcp_server.policy_approval",
+ "approve",
+ pending["request_id"],
+ "--directory",
+ str(directory),
+ ],
+ env={**os.environ, "PYTHONPATH": str(ROOT / "src")},
+ capture_output=True,
+ text=True,
+ timeout=10,
+ )
+ self.assertEqual(decision.returncode, 0, decision.stderr)
+ response = await
client.call_tool("reset_iotdb_session_policy", {})
+ self.assertTrue(
+ json.loads(response.content[0].text)["payload"]["applied"]
+ )
+
+ asyncio.run(exercise())
if __name__ == "__main__":
