From 9c22916ceeedd7a1964d4b70eb0de6d8a6171260 Mon Sep 17 00:00:00 2001 From: nobuQuartile Date: Wed, 15 Jul 2026 00:15:45 +0000 Subject: [PATCH] [FIX] ai_oca_mcp: grant internal users access to their own MCP keys The MCP endpoint switches to the key's user_id via with_user before serving tools/list and tools/call. mcp.server and mcp.server.key were only accessible to base.group_system, so a non-admin key user hit an AccessError while reading its own key and server. Grant base.group_user read access to both models and add record rules limiting internal users to their own keys (and the servers those keys belong to), while keeping full visibility for administrators. --- ai_oca_mcp/__manifest__.py | 1 + ai_oca_mcp/security/ir.model.access.csv | 2 + ai_oca_mcp/security/security.xml | 27 +++++ ai_oca_mcp/tests/test_mcp.py | 126 ++++++++++++++++++++++++ 4 files changed, 156 insertions(+) create mode 100644 ai_oca_mcp/security/security.xml diff --git a/ai_oca_mcp/__manifest__.py b/ai_oca_mcp/__manifest__.py index 884735c1..e8bbe53f 100644 --- a/ai_oca_mcp/__manifest__.py +++ b/ai_oca_mcp/__manifest__.py @@ -14,6 +14,7 @@ "data": [ "views/mcp_server_log.xml", "security/ir.model.access.csv", + "security/security.xml", "views/mcp_server_key.xml", "wizards/mcp_server_key_add.xml", "views/mcp_server.xml", diff --git a/ai_oca_mcp/security/ir.model.access.csv b/ai_oca_mcp/security/ir.model.access.csv index cf664cda..a5a53a0f 100644 --- a/ai_oca_mcp/security/ir.model.access.csv +++ b/ai_oca_mcp/security/ir.model.access.csv @@ -1,5 +1,7 @@ id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink access_mcp_server,access_mcp_server,model_mcp_server,base.group_system,1,1,1,1 access_mcp_server_key,access_mcp_server_key,model_mcp_server_key,base.group_system,1,1,1,1 +access_mcp_server_user,access_mcp_server_user,model_mcp_server,base.group_user,1,0,0,0 +access_mcp_server_key_user,access_mcp_server_key_user,model_mcp_server_key,base.group_user,1,0,0,0 access_mcp_server_key_add,access_mcp_server_key_add,model_mcp_server_key_add,base.group_system,1,1,1,1 access_mcp_server_log,access_mcp_server_log,model_mcp_server_log,base.group_system,1,0,0,0 diff --git a/ai_oca_mcp/security/security.xml b/ai_oca_mcp/security/security.xml new file mode 100644 index 00000000..01c95faf --- /dev/null +++ b/ai_oca_mcp/security/security.xml @@ -0,0 +1,27 @@ + + + + Mcp Server Key: own keys + + [('user_id', '=', user.id)] + + + + Mcp Server Key: all (admin) + + [(1, '=', 1)] + + + + Mcp Server: servers of own keys + + [('key_ids.user_id', '=', user.id)] + + + + Mcp Server: all (admin) + + [(1, '=', 1)] + + + diff --git a/ai_oca_mcp/tests/test_mcp.py b/ai_oca_mcp/tests/test_mcp.py index 15e3e454..7e2b6069 100644 --- a/ai_oca_mcp/tests/test_mcp.py +++ b/ai_oca_mcp/tests/test_mcp.py @@ -5,6 +5,7 @@ from freezegun import freeze_time +from odoo.exceptions import AccessError from odoo.tests.common import HttpCase from odoo.tools import mute_logger @@ -173,6 +174,131 @@ def test_execute_tool(self): self.assertIn("structuredContent", response["result"]) self.assertEqual(response["result"]["structuredContent"]["date"], "2024-01-01") + @mute_logger("odoo.models", "odoo.addons.base.models.ir_rule") + def test_internal_user_access(self): + """A plain internal user can read the key (and its server) it owns, + but cannot open records for which it has no key of its own.""" + group_user = self.env.ref("base.group_user") + user_a = self.env["res.users"].create( + { + "name": "MCP User A", + "login": "mcp_user_a", + "groups_id": [(6, 0, [group_user.id])], + } + ) + user_b = self.env["res.users"].create( + { + "name": "MCP User B", + "login": "mcp_user_b", + "groups_id": [(6, 0, [group_user.id])], + } + ) + self.assertFalse(user_a.has_group("base.group_system")) + # user_a owns a key on self.server + key_a = self.env["mcp.server.key"].create( + {"name": "Key A", "server_id": self.server.id, "user_id": user_a.id} + ) + # user_b owns a key on another server; user_a has no key anywhere on it + other_server = self.env["mcp.server"].create({"name": "Other Server"}) + key_b = self.env["mcp.server.key"].create( + {"name": "Key B", "server_id": other_server.id, "user_id": user_b.id} + ) + + # --- With a key of its own, the internal user can read it and its server --- + self.assertEqual(key_a.with_user(user_a).read(["name"])[0]["name"], "Key A") + self.assertEqual( + self.server.with_user(user_a).read(["name"])[0]["name"], "Test Server" + ) + self.assertIn(key_a, self.env["mcp.server.key"].with_user(user_a).search([])) + self.assertIn(self.server, self.env["mcp.server"].with_user(user_a).search([])) + + # --- Records it has no key for are filtered out and cannot be opened --- + self.assertNotIn(key_b, self.env["mcp.server.key"].with_user(user_a).search([])) + self.assertNotIn( + other_server, self.env["mcp.server"].with_user(user_a).search([]) + ) + with self.assertRaises(AccessError): + key_b.with_user(user_a).read(["name"]) + with self.assertRaises(AccessError): + other_server.with_user(user_a).read(["name"]) + + def _create_non_admin_key(self): + """Create an active key on self.server owned by a plain internal user + (base.group_user only) and return its security key.""" + user = self.env["res.users"].create( + { + "name": "MCP Plain User", + "login": "mcp_plain_user", + "groups_id": [(6, 0, [self.env.ref("base.group_user").id])], + } + ) + self.assertFalse(user.has_group("base.group_system")) + key = self.env["mcp.server.key"].create( + {"name": "Plain Key", "server_id": self.server.id, "user_id": user.id} + ) + security_key = "plain-user-security-key" + key.hashed_key = key._hash_key(security_key) + # Refresh the cached key lookup so the new key is resolvable. + self.env["mcp.server.key"]._get_mcp_server_by_key.clear_cache( + self.env["mcp.server.key"] + ) + return security_key + + def test_list_tools_non_admin_user(self): + """A key owned by a plain internal user (no ir.model access) must still + be able to list tools -- building the tool definitions must not require + Administration/Access Rights.""" + security_key = self._create_non_admin_key() + request = self.url_open( + f"/mcp/{self.server.key}", + data=json.dumps( + { + "jsonrpc": "2.0", + "id": "1", + "method": "tools/list", + } + ), + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {security_key}", + }, + ) + self.assertEqual(request.status_code, 200) + response = json.loads(request.content.decode("utf-8")) + self.assertNotIn("error", response) + self.assertIn("result", response) + self.assertEqual(1, len(response["result"]["tools"])) + self.assertEqual("get_date", response["result"]["tools"][0]["name"]) + + def test_execute_tool_non_admin_user(self): + """A generic tool must be callable through a plain internal user's key + (its model dispatch must not require ir.model read access).""" + security_key = self._create_non_admin_key() + with freeze_time("2024-01-01"): + request = self.url_open( + f"/mcp/{self.server.key}", + data=json.dumps( + { + "jsonrpc": "2.0", + "id": "1", + "method": "tools/call", + "params": { + "name": "get_date", + "arguments": {}, + }, + } + ), + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {security_key}", + }, + ) + self.assertEqual(request.status_code, 200) + response = json.loads(request.content.decode("utf-8")) + self.assertNotIn("error", response) + self.assertIn("result", response) + self.assertEqual(response["result"]["structuredContent"]["date"], "2024-01-01") + def test_url(self): self.server.key = "newkey" self.assertEqual(self.server.url, "http://127.0.0.1:8069/mcp/newkey")