fix: gate attachment uploads by project ACL

README.md

@@ -174,7 +174,9 @@ Local HTTP mode, or stdio with `--enable-writes`, exposes two additional MCP too
 
 They target only `t_projects` and the fixed field mapping in `projects` in `config.json`. A project name is required for creation; `inquiry_date` uses `YYYY-MM-DD`; progress accepts only `询价阶段`, `已报价`, `送样`, `已定点`, `SOP`, or `终止`. Attachments are added or removed incrementally—there is no replacement attachment-list operation.
 
-The same local HTTP service exposes authenticated `POST /uploads`. Submit exactly one multipart field named `file`; the default single-file limit is 25 MiB (`NOCOBASE_MCP_UPLOAD_MAX_BYTES=26214400`). A successful upload returns an `upload_id`, which callers can pass to either project tool. The upload route and `/mcp` use the same local MCP bearer token.
+Attachment upload is disabled by default. After an administrator confirms that the user's NocoBase role may create or update `t_projects`, set `NOCOBASE_MCP_ENABLE_UPLOADS=1` to expose authenticated `POST /uploads`. Submit exactly one multipart field named `file`; the default single-file limit is 25 MiB (`NOCOBASE_MCP_UPLOAD_MAX_BYTES=26214400`). A successful upload returns an `upload_id`, which callers can pass to either project tool. The upload route and `/mcp` use the same local MCP bearer token.
+
+Every upload first calls NocoBase `roles:check` with the current user's credential. The MCP follows NocoBase's global and collection-specific ACL resolution and requires `t_projects:create` or `t_projects:update` before it calls `attachments:create`. A role without project write permission receives HTTP `403` with `nb_upload_forbidden`, even if that NocoBase installation otherwise allows the generic attachment-create endpoint.
 
 Queries and writes use the saved `nb-mcp` login. For multipart uploads and attachment associations, the MCP asks `nb env info nb-mcp --json --show-secrets` for the current credential at request time. The credential is held only in memory: it is not placed in `config.json`, the MCP client configuration, logs, or subprocess arguments. If login is missing or expired, run `nb env auth nb-mcp` again.
 
@@ -189,10 +191,16 @@ nb env auth nb-mcp
 
 # Create and start the private, loopback-only user service.
 bash scripts/mcp_service.sh start
+
+# Optional: after ACL approval, change this generated 0600 file to
+# NOCOBASE_MCP_ENABLE_UPLOADS=1, then restart the service.
+${EDITOR:-vi} .runtime/nocobase-crm-mcp.env
+bash scripts/mcp_service.sh start
+
 curl -X POST -H "Authorization: Bearer <mcp-token>" -F "file=@quote.pdf" http://127.0.0.1:10087/uploads
 ```
 
-The NocoBase login belongs to the user's own `nb` environment, not `.runtime/nocobase-crm-mcp.env`. The MCP bearer token protects only the loopback HTTP endpoint. An uploaded attachment that is not yet associated with a project is intentionally retained; this release does not auto-clean orphan attachments. It also provides no delete operation for projects or attachments.
+The NocoBase login belongs to the user's own `nb` environment, not `.runtime/nocobase-crm-mcp.env`. The MCP bearer token protects only the loopback HTTP endpoint. An authorized upload that is not yet associated with a project is intentionally retained; this release does not auto-clean orphan attachments. Enable uploads only for approved project writers and monitor unassociated attachments. The MCP provides no delete operation for projects or attachments.
 
 ## Permissions
 
@@ -217,7 +225,7 @@ For either write-enabled mode, the logged-in user additionally needs:
 - `attachments`: upload/create
 - `t_projects.f_projects_attachments`: add and remove attachment relations
 
-Do not grant project or attachment delete permissions. This server does not offer deletes, and it retains uploaded but unassociated attachments.
+Keep `NOCOBASE_MCP_ENABLE_UPLOADS=0` for read-only users and project roles without create/update access. Before enabling uploads, test the exact role with both `roles:check` and a real denied/allowed upload case. Do not grant project or attachment delete permissions solely for MCP use; this server does not expose deletes, and operators should handle any retained orphan attachments through an audited administrative cleanup process.
 
 ## Local HTTP MCP (optional)
 
@@ -225,6 +233,7 @@ HTTP mode uses streamable HTTP with a separate local bearer token for MCP access
 
 ```bash
 export NOCOBASE_MCP_TOKEN='<long-random-token>'
+export NOCOBASE_MCP_ENABLE_UPLOADS=0
 uv run nocobase-crm-mcp --transport streamable-http --config config.json
 ```
 

docs/LOCAL_DELIVERY.md

@@ -74,15 +74,18 @@ nb env auth nb-mcp
 
 默认只读。如需项目创建和更新，在参数末尾加入 `--enable-writes`。这些写工具只允许预先定义的项目字段，不提供删除或任意资源访问能力。
 
-附件上传需要可选的本机 HTTP 模式。它默认只监听 `127.0.0.1:10087`，不得暴露到局域网或公网：
+附件上传需要可选的本机 HTTP 模式。HTTP 服务默认只监听 `127.0.0.1:10087`；上传路由本身也默认关闭，不得暴露到局域网或公网：
 
 ```bash
 NOCOBASE_MCP_TOKEN='<随机长字符串>' \
+NOCOBASE_MCP_ENABLE_UPLOADS=0 \
   uv run nocobase-crm-mcp --transport streamable-http --config config.json
 ```
 
 Linux 用户也可以执行 `bash scripts/mcp_service.sh start` 安装用户级 systemd 服务。
 
+只有管理员确认该用户角色拥有 `t_projects` 的 create 或 update 后，才能在 `.runtime/nocobase-crm-mcp.env` 中设置 `NOCOBASE_MCP_ENABLE_UPLOADS=1` 并重新执行 `bash scripts/mcp_service.sh start`。即使已显式启用，MCP 仍会在每次上传前调用当前账号的 `roles:check`；无项目写权限时返回 `403 / nb_upload_forbidden`，不会调用 `attachments:create`。
+
 ## 权限验收
 
 交付前至少使用两个权限不同的测试账号验证：
@@ -91,8 +94,11 @@ Linux 用户也可以执行 `bash scripts/mcp_service.sh start` 安装用户级
 2. 使用相同参数调用 `list_customers`、`list_contacts`、`list_followups` 和 `list_projects`。
 3. 确认各自只能看到 NocoBase UI 中本人有权查看的数据。
 4. 对无项目写权限的账号调用 `create_project`，确认 NocoBase 拒绝。
-5. 对有写权限的账号创建测试项目，确认审计字段记录的是该用户，而不是共享服务账号。
-6. 退出或使 Token 过期后，确认 MCP 返回 `nb_auth_failed`，且不会降级使用其他凭证。
+5. 显式启用上传后，对无项目写权限的账号上传一个小文件，确认返回 `nb_upload_forbidden`，并用管理员确认附件记录数没有增加。
+6. 对有写权限的账号创建测试项目并完成上传、关联和移除，确认审计字段记录的是该用户，而不是共享服务账号。
+7. 退出或使 Token 过期后，确认 MCP 返回 `nb_auth_failed`，且不会降级使用其他凭证。
+
+修改测试账号角色后，如果 `roles:check` 仍返回旧角色，先由管理员执行 `nb api app clear-cache`，再重新执行 `nb env auth <环境名>`。这是服务端角色缓存刷新，不应通过共享 Token 绕过。
 
 ## 凭证安全
 
@@ -102,6 +108,7 @@ Linux 用户也可以执行 `bash scripts/mcp_service.sh start` 安装用户级
 - 不要共享操作系统账号运行 MCP；本机权限隔离依赖每位用户拥有独立的 `nb` 登录存储。
 - `NOCOBASE_MCP_API_TOKEN` 只为旧部署兼容保留，不用于终端用户交付。
 - `include_raw=true` 默认被 MCP 拒绝。只有在管理员审核 `config.json` 字段和 AI 数据使用政策后，才可在该用户本机设置 `NOCOBASE_MCP_ENABLE_RAW_DATA=1`。
+- `POST /uploads` 默认不存在。只有在管理员完成角色验收后才设置 `NOCOBASE_MCP_ENABLE_UPLOADS=1`；MCP 的项目写权限门禁是纵深防御，NocoBase ACL 仍是最终权限边界。
 
 ## 更新、回滚与恢复
 
@@ -148,4 +155,5 @@ bash scripts/mcp_service.sh start
 - `nb_auth_failed`：执行 `nb env auth nb-mcp`。
 - `NO_PERMISSIONS`：由 NocoBase 管理员调整该用户的角色、字段或行级数据权限。
 - 附件返回 `nb_upload_unauthorized`：重新登录后重试；不要把其他用户的 Token 写入环境变量。
+- 附件返回 `nb_upload_forbidden`：当前角色没有项目 create/update；不要为了绕过门禁而改用共享高权限账号。
 - 环境连接错误：执行 `nb env info nb-mcp`，确认 `apiBaseUrl` 指向目标环境的 `/api`。

docs/superpowers/specs/2026-07-12-project-write-design.md

@@ -33,7 +33,7 @@
 
 ## HTTP 上传接口
 
-新增 `POST /uploads`，与 `/mcp` 部署在同一个 HTTP 服务中，并复用现有 MCP Bearer Token 中间件。
+新增可选的 `POST /uploads`，与 `/mcp` 部署在同一个 HTTP 服务中，并复用现有 MCP Bearer Token 中间件。交付加固后该路由默认关闭；只有设置 `NOCOBASE_MCP_ENABLE_UPLOADS=1` 才注册。
 
 请求必须为 `multipart/form-data`，每次仅接收一个名为 `file` 的文件字段。默认单文件上限为 25 MiB。首版不额外限制 MIME 类型；部署者可在配置中收紧允许的类型。
 
@@ -54,7 +54,7 @@
 }
 ```
 
-`upload_id` 是 NocoBase `attachments.id`。上传成功但尚未关联到项目的附件会保留；首版不运行自动清理任务，以避免误删业务文件。
+`upload_id` 是 NocoBase `attachments.id`。上传前必须以当前用户凭证调用 `roles:check`，并按 NocoBase 的全局或表级 ACL 解析确认 `t_projects:create/update` 至少一项可用。上传成功但尚未关联到项目的附件会保留；首版不运行自动清理任务，以避免误删业务文件。
 
 ## MCP 工具
 
@@ -114,7 +114,7 @@ update_project(
 - `NOCOBASE_MCP_UPLOAD_MAX_BYTES`（默认 `26214400`）
 - 可选 MIME 白名单环境变量或配置项
 
-服务角色必须拥有：`t_projects` 的 create/update、`attachments` 的上传/create，以及 `t_projects.f_projects_attachments` 的 add/remove 权限。README 和 systemd 服务配置必须明确这些新增权限与密钥配置。
+服务角色必须拥有：`t_projects` 的 create/update、`attachments` 的上传/create，以及 `t_projects.f_projects_attachments` 的 add/remove 权限。README 和 systemd 服务配置必须明确这些新增权限与密钥配置；只读用户保持 `NOCOBASE_MCP_ENABLE_UPLOADS=0`。
 
 ## 验证
 

nocobase_crm_mcp/nocobase_http.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Callable, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import Any
 from urllib.parse import quote
@@ -24,6 +24,7 @@ class NocoBaseHttpError(RuntimeError):
 class NocoBaseHttpClient:
     """Small synchronous client for the fixed NocoBase attachment endpoints."""
 
+    _PROJECT_RESOURCE = "t_projects"
     _ATTACHMENT_FIELD = "t_projects.f_projects_attachments"
 
     def __init__(
@@ -50,6 +51,7 @@ class NocoBaseHttpClient:
 
     def upload_attachment(self, path: Path, filename: str, mimetype: str | None) -> dict[str, Any]:
         api_base_url, api_token = self._credentials()
+        self._ensure_project_write_access(api_base_url, api_token)
         with Path(path).open("rb") as handle:
             response = self._client.post(
                 f"{api_base_url}/attachments:create",
@@ -63,6 +65,27 @@ class NocoBaseHttpClient:
             raise NocoBaseHttpError(response.status_code, "NocoBase response did not contain an attachment object")
         return data
 
+    def _ensure_project_write_access(self, api_base_url: str, api_token: str) -> None:
+        response = self._client.get(
+            f"{api_base_url}/roles:check",
+            headers={"Authorization": f"Bearer {api_token}"},
+        )
+        payload = self._json_response(response, api_token)
+        role_data = payload.get("data")
+        if not isinstance(role_data, Mapping):
+            raise NocoBaseHttpError(502, "NocoBase roles:check response did not contain role data")
+
+        effective_role = dict(role_data)
+        meta = payload.get("meta")
+        data_sources = meta.get("dataSources") if isinstance(meta, Mapping) else None
+        main_role = data_sources.get("main") if isinstance(data_sources, Mapping) else None
+        if isinstance(main_role, Mapping):
+            effective_role.update(main_role)
+
+        if any(_role_allows_action(effective_role, self._PROJECT_RESOURCE, action) for action in ("create", "update")):
+            return
+        raise NocoBaseHttpError(403, "Current NocoBase role cannot create or update projects")
+
     def add_attachment_ids(self, project_id: Any, attachment_ids: Sequence[Any]) -> None:
         self._post_attachment_ids(project_id, attachment_ids, "add")
 
@@ -125,6 +148,32 @@ def _project_id_segment(project_id: Any) -> str:
     return quote(validate_project_id(project_id), safe="")
 
 
+def _role_allows_action(role_data: Mapping[str, Any], resource: str, action: str) -> bool:
+    if role_data.get("allowAll") is True:
+        return True
+
+    aliases = role_data.get("actionAlias")
+    resolved_action = aliases.get(action, action) if isinstance(aliases, Mapping) else action
+    resources = role_data.get("resources")
+    if isinstance(resources, Sequence) and not isinstance(resources, (str, bytes)) and resource in resources:
+        actions = role_data.get("actions")
+        if not isinstance(actions, Mapping):
+            return False
+        for action_path in (f"{resource}:{resolved_action}", f"{resource}:{action}"):
+            if action_path in actions and actions[action_path] is not None and actions[action_path] is not False:
+                return True
+        return False
+
+    strategy = role_data.get("strategy")
+    strategy_actions = strategy.get("actions") if isinstance(strategy, Mapping) else None
+    if not isinstance(strategy_actions, Sequence) or isinstance(strategy_actions, (str, bytes)):
+        return False
+    return any(
+        isinstance(strategy_action, str) and strategy_action.split(":", 1)[0] == resolved_action
+        for strategy_action in strategy_actions
+    )
+
+
 def validate_project_id(project_id: Any) -> str:
     if isinstance(project_id, bool) or not isinstance(project_id, (str, int)):
         raise ValueError("project_id must be a scalar identifier")

nocobase_crm_mcp/server.py

@@ -28,6 +28,7 @@ DEFAULT_HTTP_PATH = "/mcp"
 DEFAULT_TOKEN_ENV = "NOCOBASE_MCP_TOKEN"
 ENABLE_CONFIG_TOOLS_ENV = "NOCOBASE_MCP_ENABLE_CONFIG_TOOLS"
 ENABLE_RAW_DATA_ENV = "NOCOBASE_MCP_ENABLE_RAW_DATA"
+ENABLE_UPLOADS_ENV = "NOCOBASE_MCP_ENABLE_UPLOADS"
 UPLOAD_API_BASE_URL_ENV = "NOCOBASE_MCP_UPLOAD_API_BASE_URL"
 UPLOAD_JWT_ENV = "NOCOBASE_MCP_UPLOAD_JWT"
 UPLOAD_MAX_BYTES_ENV = "NOCOBASE_MCP_UPLOAD_MAX_BYTES"
@@ -716,7 +717,9 @@ def run_http_server(
         allowed_hosts=allowed_hosts,
         allowed_origins=allowed_origins,
     )
-    upload_endpoint = UploadEndpoint(http_client, max_bytes=max_bytes, temp_dir=Path(tempfile.gettempdir()))
+    upload_endpoint = None
+    if os.environ.get(ENABLE_UPLOADS_ENV, "").strip() == "1":
+        upload_endpoint = UploadEndpoint(http_client, max_bytes=max_bytes, temp_dir=Path(tempfile.gettempdir()))
     app = create_http_app(server, token, upload_endpoint=upload_endpoint)
     uvicorn.run(app, host=host, port=port, log_level="info")
 

nocobase_crm_mcp/uploads.py

@@ -84,8 +84,15 @@ class UploadEndpoint:
                 return _error_response("nb_upload_failed", "NocoBase attachment upload failed", 502)
             return JSONResponse({"upload_id": attachment_id, "filename": filename, "size": size, "mimetype": mimetype})
         except NocoBaseHttpError as error:
-            code = "nb_upload_unauthorized" if error.authentication_failed else "nb_upload_failed"
-            return _error_response(code, "NocoBase attachment upload failed", 502)
+            if error.authentication_failed:
+                return _error_response("nb_upload_unauthorized", "NocoBase attachment upload failed", 502)
+            if error.status_code == 403:
+                return _error_response(
+                    "nb_upload_forbidden",
+                    "The current NocoBase role cannot upload project attachments",
+                    403,
+                )
+            return _error_response("nb_upload_failed", "NocoBase attachment upload failed", 502)
         except Exception:
             return _error_response("nb_upload_failed", "NocoBase attachment upload failed", 502)
         finally:

scripts/manage_mcp_service.py

@@ -28,6 +28,7 @@ DEFAULT_NB_BIN = "nb"
 DEFAULT_TOKEN_BYTES = 32
 DEFAULT_UV_CACHE_DIR = "/tmp/uv-cache"
 DEFAULT_UPLOAD_MAX_BYTES = 25 * 1024 * 1024
+ENABLE_UPLOADS_ENV = "NOCOBASE_MCP_ENABLE_UPLOADS"
 API_BASE_URL_ENV = "NOCOBASE_MCP_API_BASE_URL"
 API_TOKEN_ENV = "NOCOBASE_MCP_API_TOKEN"
 LEGACY_API_BASE_URL_ENV = "NOCOBASE_MCP_UPLOAD_API_BASE_URL"
@@ -41,6 +42,7 @@ MANAGED_ENV_NAMES = frozenset(
         "NOCOBASE_MCP_ENV",
         "NOCOBASE_MCP_CONFIG",
         "NOCOBASE_MCP_NB_BIN",
+        ENABLE_UPLOADS_ENV,
         "NOCOBASE_MCP_UPLOAD_MAX_BYTES",
     }
 )
@@ -150,6 +152,7 @@ def write_token_env_file(
     config_path: str = DEFAULT_CONFIG_PATH,
     nb_bin: str | Path = DEFAULT_NB_BIN,
 ) -> None:
+    uploads_enabled = _saved_uploads_setting(env_path)
     preserved_lines = _operator_env_lines(env_path)
     env_path.parent.mkdir(parents=True, exist_ok=True)
     env_path.parent.chmod(0o700)
@@ -162,6 +165,7 @@ def write_token_env_file(
         env_file.write(f"NOCOBASE_MCP_ENV={nb_env}\n")
         env_file.write(f"NOCOBASE_MCP_CONFIG={config_path}\n")
         env_file.write(f"NOCOBASE_MCP_NB_BIN={nb_bin}\n")
+        env_file.write(f"{ENABLE_UPLOADS_ENV}={uploads_enabled}\n")
         env_file.write(f"NOCOBASE_MCP_UPLOAD_MAX_BYTES={DEFAULT_UPLOAD_MAX_BYTES}\n")
         if preserved_lines:
             env_file.write("\n")
@@ -170,6 +174,16 @@ def write_token_env_file(
     env_path.chmod(0o600)
 
 
+def _saved_uploads_setting(env_path: Path) -> str:
+    if not env_path.exists():
+        return "0"
+    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
+        key, separator, value = raw_line.strip().partition("=")
+        if separator and key.strip() == ENABLE_UPLOADS_ENV:
+            return "1" if value.strip() == "1" else "0"
+    return "0"
+
+
 def _operator_env_lines(env_path: Path) -> list[str]:
     if not env_path.exists():
         return []
@@ -456,6 +470,8 @@ def service_status(*, project_root: Path) -> str:
         lines.append(f"MCP URL: http://{display_host}:{port}{path}")
         lines.append(f"nb binary: {nb_bin}")
         lines.append(f"NocoBase CLI env: {nb_env} (authenticate with `nb env auth {nb_env}`)")
+        uploads_enabled = values.get(ENABLE_UPLOADS_ENV) == "1"
+        lines.append(f"Attachment uploads: {'enabled' if uploads_enabled else 'disabled (safe default)'}")
         canonical_configured = bool(values.get(API_BASE_URL_ENV) and values.get(API_TOKEN_ENV))
         legacy_configured = bool(values.get(LEGACY_API_BASE_URL_ENV) and values.get(LEGACY_API_TOKEN_ENV))
         if canonical_configured:

tests/test_mcp_service.py

@@ -73,6 +73,7 @@ class McpServiceConfigTests(unittest.TestCase):
             self.assertIn("NOCOBASE_MCP_ENV=nb-mcp\n", content)
             self.assertIn("NOCOBASE_MCP_CONFIG=config.json\n", content)
             self.assertIn("NOCOBASE_MCP_NB_BIN=/opt/node/bin/nb\n", content)
+            self.assertIn("NOCOBASE_MCP_ENABLE_UPLOADS=0\n", content)
             self.assertEqual(stat.S_IMODE(os.stat(env_path).st_mode), 0o600)
             self.assertEqual(stat.S_IMODE(os.stat(env_path.parent).st_mode), 0o700)
 
@@ -133,6 +134,7 @@ class McpServiceConfigTests(unittest.TestCase):
 
             content = env_path.read_text(encoding="utf-8")
             self.assertIn("NOCOBASE_MCP_UPLOAD_MAX_BYTES=26214400\n", content)
+            self.assertIn("NOCOBASE_MCP_ENABLE_UPLOADS=0\n", content)
             self.assertNotIn("NOCOBASE_MCP_API_TOKEN", content)
             self.assertNotIn("NOCOBASE_MCP_API_BASE_URL", content)
             self.assertNotIn("NOCOBASE_MCP_UPLOAD_JWT", content)
@@ -157,6 +159,22 @@ class McpServiceConfigTests(unittest.TestCase):
             self.assertIn("NOCOBASE_MCP_API_BASE_URL=https://nocobase.example/api\n", content)
             self.assertIn("NOCOBASE_MCP_API_TOKEN=operator-api-key\n", content)
 
+    def test_token_rotation_preserves_explicit_upload_opt_in(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            env_path = Path(tmp) / "mcp.env"
+            manage_mcp_service.write_token_env_file(env_path, "old-token")
+            content = env_path.read_text(encoding="utf-8").replace(
+                "NOCOBASE_MCP_ENABLE_UPLOADS=0",
+                "NOCOBASE_MCP_ENABLE_UPLOADS=1",
+            )
+            env_path.write_text(content, encoding="utf-8")
+
+            manage_mcp_service.write_token_env_file(env_path, "new-token")
+
+            content = env_path.read_text(encoding="utf-8")
+            self.assertIn("NOCOBASE_MCP_ENABLE_UPLOADS=1\n", content)
+            self.assertEqual(content.count("NOCOBASE_MCP_ENABLE_UPLOADS="), 1)
+
     def test_render_http_config_prints_url_and_authorization_header(self) -> None:
         output = manage_mcp_service.render_remote_config(
             hosts=["192.168.1.10"],
@@ -412,6 +430,7 @@ class McpServiceConfigTests(unittest.TestCase):
         self.assertIn("Env file:", output)
         self.assertIn("http://127.0.0.1:10087/mcp", output)
         self.assertIn("NocoBase CLI env: nb-mcp", output)
+        self.assertIn("Attachment uploads: disabled (safe default)", output)
         self.assertIn("HTTP probe: reachable", output)
 
     def test_service_status_reports_canonical_credentials_without_exposing_them(self) -> None:

tests/test_nocobase_http.py

@@ -32,11 +32,15 @@ class ProjectConfigTests(unittest.TestCase):
 
 class NocoBaseHttpClientTests(unittest.TestCase):
     def test_upload_posts_file_with_attachment_field(self) -> None:
-        seen: dict[str, str] = {}
+        seen: list[tuple[str, str]] = []
 
         def handler(request: httpx.Request) -> httpx.Response:
-            seen["url"] = str(request.url)
-            seen["auth"] = request.headers["authorization"]
+            seen.append((str(request.url), request.headers["authorization"]))
+            if request.url.path.endswith("/roles:check"):
+                return httpx.Response(
+                    200,
+                    json={"data": {"strategy": {"actions": ["view", "create"]}, "resources": []}},
+                )
             return httpx.Response(200, json={"data": {"id": 42}})
 
         with TemporaryDirectory() as directory:
@@ -46,8 +50,87 @@ class NocoBaseHttpClientTests(unittest.TestCase):
 
             self.assertEqual(client.upload_attachment(path, "quote.pdf", "application/pdf")["id"], 42)
 
-        self.assertTrue(seen["url"].endswith("/attachments:create?attachmentField=t_projects.f_projects_attachments"))
-        self.assertEqual(seen["auth"], "Bearer jwt")
+        self.assertTrue(seen[0][0].endswith("/roles:check"))
+        self.assertTrue(seen[1][0].endswith("/attachments:create?attachmentField=t_projects.f_projects_attachments"))
+        self.assertEqual([auth for _url, auth in seen], ["Bearer jwt", "Bearer jwt"])
+
+    def test_upload_accepts_collection_specific_project_update_permission(self) -> None:
+        requests: list[httpx.Request] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            requests.append(request)
+            if request.url.path.endswith("/roles:check"):
+                return httpx.Response(
+                    200,
+                    json={
+                        "data": {
+                            "strategy": {"actions": ["view:own"]},
+                            "resources": ["t_projects"],
+                            "actions": {"t_projects:update": {}},
+                        }
+                    },
+                )
+            return httpx.Response(200, json={"data": {"id": 42}})
+
+        with TemporaryDirectory() as directory:
+            path = Path(directory) / "quote.pdf"
+            path.write_bytes(b"pdf")
+            client = NocoBaseHttpClient("https://nb.example/api", "jwt", transport=httpx.MockTransport(handler))
+
+            self.assertEqual(client.upload_attachment(path, "quote.pdf", "application/pdf")["id"], 42)
+
+        self.assertEqual(len(requests), 2)
+
+    def test_upload_rejects_role_without_project_write_before_attachment_create(self) -> None:
+        requests: list[httpx.Request] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            requests.append(request)
+            return httpx.Response(
+                200,
+                json={"data": {"strategy": {"actions": ["view:own", "trigger:own"]}, "resources": []}},
+            )
+
+        with TemporaryDirectory() as directory:
+            path = Path(directory) / "quote.pdf"
+            path.write_bytes(b"pdf")
+            client = NocoBaseHttpClient("https://nb.example/api", "jwt", transport=httpx.MockTransport(handler))
+
+            with self.assertRaises(NocoBaseHttpError) as caught:
+                client.upload_attachment(path, "quote.pdf", "application/pdf")
+
+        self.assertEqual(caught.exception.status_code, 403)
+        self.assertFalse(caught.exception.authentication_failed)
+        self.assertEqual([request.url.path for request in requests], ["/api/roles:check"])
+
+    def test_upload_accepts_root_role_and_main_data_source_acl(self) -> None:
+        role_payloads = (
+            {"data": {"allowAll": True}},
+            {
+                "data": {"strategy": {"actions": ["view:own"]}, "resources": []},
+                "meta": {
+                    "dataSources": {
+                        "main": {
+                            "strategy": {"actions": ["view", "update:own"]},
+                            "resources": [],
+                        }
+                    }
+                },
+            },
+        )
+
+        for role_payload in role_payloads:
+            with self.subTest(role_payload=role_payload), TemporaryDirectory() as directory:
+                path = Path(directory) / "quote.pdf"
+                path.write_bytes(b"pdf")
+
+                def handler(request: httpx.Request) -> httpx.Response:
+                    if request.url.path.endswith("/roles:check"):
+                        return httpx.Response(200, json=role_payload)
+                    return httpx.Response(200, json={"data": {"id": 42}})
+
+                client = NocoBaseHttpClient("https://nb.example/api", "jwt", transport=httpx.MockTransport(handler))
+                self.assertEqual(client.upload_attachment(path, "quote.pdf", "application/pdf")["id"], 42)
 
     def test_attachment_associations_post_ids_to_fixed_relation(self) -> None:
         requests: list[httpx.Request] = []

tests/test_server.py

@@ -271,6 +271,21 @@ class ServerTests(unittest.TestCase):
 
         self.assertEqual(run_asgi_request(app, path="/uploads")[0]["status"], 401)
 
+    def test_upload_route_is_not_exposed_without_an_endpoint(self) -> None:
+        if FastMCP is None:
+            self.skipTest("MCP SDK is not installed")
+
+        server = create_server(crm_service=FakeCrmService(), readonly=True)
+        app = create_http_app(server, "secret")
+
+        async def request_upload() -> httpx.Response:
+            async with httpx.AsyncClient(
+                transport=httpx.ASGITransport(app=app), base_url="http://testserver"
+            ) as client:
+                return await client.post("/uploads", headers={"Authorization": "Bearer secret"})
+
+        self.assertEqual(asyncio.run(request_upload()).status_code, 404)
+
     def test_upload_route_accepts_the_matching_bearer_token(self) -> None:
         if FastMCP is None:
             self.skipTest("MCP SDK is not installed")
@@ -428,6 +443,60 @@ class ServerTests(unittest.TestCase):
             log_level="info",
         )
 
+    def test_http_upload_route_is_disabled_by_default(self) -> None:
+        with patch.dict("os.environ", {"NOCOBASE_MCP_TOKEN": "secret"}, clear=True):
+            with (
+                patch.object(server_module, "_make_write_services") as make_services,
+                patch.object(server_module, "create_server") as create_server_mock,
+                patch.object(server_module, "UploadEndpoint") as upload_endpoint,
+                patch.object(server_module, "create_http_app", return_value="app") as create_http_app_mock,
+                patch("uvicorn.run"),
+            ):
+                make_services.return_value = ("crm", "projects", "http")
+                run_http_server(
+                    config_path=None,
+                    host=DEFAULT_HTTP_HOST,
+                    port=DEFAULT_HTTP_PORT,
+                    token_env="NOCOBASE_MCP_TOKEN",
+                    streamable_http_path=DEFAULT_HTTP_PATH,
+                    allowed_hosts=None,
+                    allowed_origins=None,
+                )
+
+        upload_endpoint.assert_not_called()
+        create_http_app_mock.assert_called_once_with(create_server_mock.return_value, "secret", upload_endpoint=None)
+
+    def test_http_upload_route_requires_explicit_local_opt_in(self) -> None:
+        with patch.dict(
+            "os.environ",
+            {"NOCOBASE_MCP_TOKEN": "secret", "NOCOBASE_MCP_ENABLE_UPLOADS": "1"},
+            clear=True,
+        ):
+            with (
+                patch.object(server_module, "_make_write_services") as make_services,
+                patch.object(server_module, "create_server") as create_server_mock,
+                patch.object(server_module, "UploadEndpoint") as upload_endpoint,
+                patch.object(server_module, "create_http_app", return_value="app") as create_http_app_mock,
+                patch("uvicorn.run"),
+            ):
+                make_services.return_value = ("crm", "projects", "http")
+                run_http_server(
+                    config_path=None,
+                    host=DEFAULT_HTTP_HOST,
+                    port=DEFAULT_HTTP_PORT,
+                    token_env="NOCOBASE_MCP_TOKEN",
+                    streamable_http_path=DEFAULT_HTTP_PATH,
+                    allowed_hosts=None,
+                    allowed_origins=None,
+                )
+
+        upload_endpoint.assert_called_once()
+        create_http_app_mock.assert_called_once_with(
+            create_server_mock.return_value,
+            "secret",
+            upload_endpoint=upload_endpoint.return_value,
+        )
+
     def test_config_template_tool_is_opt_in(self) -> None:
         if FastMCP is None:
             self.skipTest("MCP SDK is not installed")

tests/test_uploads.py

@@ -176,7 +176,6 @@ class UploadEndpointTests(unittest.IsolatedAsyncioTestCase):
     async def test_upload_maps_upstream_errors_and_removes_temp_file(self) -> None:
         cases = (
             (NocoBaseHttpError(401, "upstream failure"), "nb_upload_unauthorized"),
-            (NocoBaseHttpError(403, "permission denied"), "nb_upload_failed"),
             (NocoBaseHttpError(403, "expired", authentication_failed=True), "nb_upload_unauthorized"),
             (NocoBaseHttpError(500, "upstream failure"), "nb_upload_failed"),
         )
@@ -199,6 +198,20 @@ class UploadEndpointTests(unittest.IsolatedAsyncioTestCase):
                 self.assertEqual(response.json()["error"]["code"], error_code)
                 self.assertEqual(list(Path(temp_dir).iterdir()), [])
 
+    async def test_upload_maps_project_write_denial_to_forbidden(self) -> None:
+        http_client = FakeHttpClient()
+        http_client.error = NocoBaseHttpError(403, "permission denied")
+        with TemporaryDirectory() as temp_dir:
+            async with await self._client_for(http_client, temp_dir) as client:
+                response = await client.post(
+                    "/uploads",
+                    headers=AUTH,
+                    files={"file": ("quote.pdf", b"pdf", "application/pdf")},
+                )
+
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.json()["error"]["code"], "nb_upload_forbidden")
+
     async def test_upload_requires_post_and_multipart_form_data(self) -> None:
         http_client = FakeHttpClient()
         with TemporaryDirectory() as temp_dir: