fix: restrict agent permission (#20)

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 @@
 只有需要 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 @@
 `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 @@
 [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 @@
 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 @@
     }
 )
 
+# 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 @@
         "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 @@
         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:
-    with _SESSION_POLICY_LOCK:
-        value = _SESSION_POLICY.get(name)
-    if value not in (None, ""):
-        return value
+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:
+        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 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 _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 @@
             + ", ".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 @@
     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 @@
     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 @@
 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 @@
 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 @@
   - `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 @@
 
 ## 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 @@
 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 @@
 }
 ```
 
-or:
+Widening within the startup deployment ceiling can be requested with:
 
 ```json
 {
@@ -325,6 +329,14 @@
 }
 ```
 
+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 @@
                 "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 @@
             )
 
         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 @@
             )
 
         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 @@
         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 @@
         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 @@
                 "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 @@
             )
 
         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 @@
                 "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 @@
                 "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 @@
                 "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 @@
                 "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 @@
     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 @@
                 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),
-        ):
-            set_session_policy({"TIMESEEK_MCP_PERMISSION_ENFORCEMENT": "strict"})
-            snapshot = set_session_policy(preset="readonly", replace=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"], {})
 
-            self.assertTrue(strict_permission_enforcement())
-            self.assertEqual(
-                snapshot["session_policy"]["TIMESEEK_MCP_PERMISSION_ENFORCEMENT"],
-                "strict",
+    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},
+        ):
+            with self.assertRaises(PermissionError):
+                set_session_policy(policy)
+
+    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__":