From b51be25e2e5ad3b886a47e1159f112fdb8de510d Mon Sep 17 00:00:00 2001 From: nac Date: Mon, 13 Jul 2026 02:24:58 +0000 Subject: [PATCH] fix: reject dot project ids --- nocobase_crm_mcp/nocobase_http.py | 13 ++++++++++--- nocobase_crm_mcp/projects.py | 11 +++++++++-- tests/test_nocobase_http.py | 11 +++++++++++ tests/test_projects.py | 10 ++++++++++ 4 files changed, 40 insertions(+), 5 deletions(-) diff --git a/nocobase_crm_mcp/nocobase_http.py b/nocobase_crm_mcp/nocobase_http.py index b0eb21b..020a2fc 100644 --- a/nocobase_crm_mcp/nocobase_http.py +++ b/nocobase_crm_mcp/nocobase_http.py @@ -86,8 +86,15 @@ class NocoBaseHttpClient: def _project_id_segment(project_id: Any) -> str: + return quote(validate_project_id(project_id), safe="") + + +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") - if isinstance(project_id, str) and not project_id.strip(): - raise ValueError("project_id must not be blank") - return quote(str(project_id), safe="") + if isinstance(project_id, str): + if not project_id.strip(): + raise ValueError("project_id must not be blank") + if project_id in {".", ".."}: + raise ValueError("project_id must not be a dot path segment") + return str(project_id) diff --git a/nocobase_crm_mcp/projects.py b/nocobase_crm_mcp/projects.py index e2ba1af..f71c280 100644 --- a/nocobase_crm_mcp/projects.py +++ b/nocobase_crm_mcp/projects.py @@ -5,7 +5,7 @@ from datetime import date from typing import Any from .config import ProjectConfig -from .nocobase_http import NocoBaseHttpError +from .nocobase_http import NocoBaseHttpError, validate_project_id def _configured_scalar_fields(config: ProjectConfig) -> tuple[str, ...]: @@ -97,7 +97,7 @@ class ProjectService: remove_attachment_ids: Sequence[Any] | None = None, ) -> dict[str, Any]: try: - _validate_identifier(project_id, "project_id") + _validate_project_id(project_id) values = self._project_values( name=name, customer_id=customer_id, @@ -223,6 +223,13 @@ def _validate_identifier(value: Any, name: str) -> None: raise _InvalidProjectInput(f"{name} must not be blank") +def _validate_project_id(project_id: Any) -> None: + try: + validate_project_id(project_id) + except ValueError as error: + raise _InvalidProjectInput(str(error)) from error + + def _attachment_error(project_id: Any, error: Exception) -> dict[str, Any]: if isinstance(error, NocoBaseHttpError): if error.status_code == 404: diff --git a/tests/test_nocobase_http.py b/tests/test_nocobase_http.py index c033c0b..e7c6f59 100644 --- a/tests/test_nocobase_http.py +++ b/tests/test_nocobase_http.py @@ -92,6 +92,17 @@ class NocoBaseHttpClientTests(unittest.TestCase): with self.assertRaisesRegex(ValueError, "project_id"): client.add_attachment_ids({"id": 7}, [8]) + def test_attachment_association_rejects_dot_path_project_ids_before_request(self) -> None: + client = NocoBaseHttpClient( + "https://nb.example/api", + "jwt", + transport=httpx.MockTransport(lambda request: self.fail("request must not be sent")), + ) + + for project_id in (".", ".."): + with self.subTest(project_id=project_id), self.assertRaisesRegex(ValueError, "project_id"): + client.add_attachment_ids(project_id, [8]) + def test_rejects_blank_upload_credentials(self) -> None: with self.assertRaises(ValueError): NocoBaseHttpClient(" ", "jwt") diff --git a/tests/test_projects.py b/tests/test_projects.py index 9095b3b..a99f075 100644 --- a/tests/test_projects.py +++ b/tests/test_projects.py @@ -145,6 +145,16 @@ class ProjectServiceTests(unittest.TestCase): self.assertEqual(self.nb.update_calls, []) self.assertEqual(self.http.add_calls, []) + def test_update_rejects_dot_path_project_id_before_any_client_call(self) -> None: + for project_id in (".", ".."): + with self.subTest(project_id=project_id): + self.setUp() + result = self.service.update_project(project_id, name="x", add_attachment_ids=[8]) + + self.assertEqual(result["error"]["code"], "invalid_project_input") + self.assertEqual(self.nb.update_calls, []) + self.assertEqual(self.http.add_calls, []) + def test_create_rejects_nested_attachment_id_before_any_client_call(self) -> None: result = self.service.create_project(name="x", attachment_ids=[{"id": 8}])