[16.0][FIX] ai_oca_mcp: grant internal users access to their own MCP keys

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",

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

ai_oca_mcp/security/security.xml

@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<odoo noupdate="1">
+    <record id="mcp_server_key_rule_user" model="ir.rule">
+        <field name="name">Mcp Server Key: own keys</field>
+        <field name="model_id" ref="model_mcp_server_key" />
+        <field name="domain_force">[('user_id', '=', user.id)]</field>
+        <field name="groups" eval="[Command.link(ref('base.group_user'))]" />
+    </record>
+    <record id="mcp_server_key_rule_admin" model="ir.rule">
+        <field name="name">Mcp Server Key: all (admin)</field>
+        <field name="model_id" ref="model_mcp_server_key" />
+        <field name="domain_force">[(1, '=', 1)]</field>
+        <field name="groups" eval="[Command.link(ref('base.group_system'))]" />
+    </record>
+    <record id="mcp_server_rule_user" model="ir.rule">
+        <field name="name">Mcp Server: servers of own keys</field>
+        <field name="model_id" ref="model_mcp_server" />
+        <field name="domain_force">[('key_ids.user_id', '=', user.id)]</field>
+        <field name="groups" eval="[Command.link(ref('base.group_user'))]" />
+    </record>
+    <record id="mcp_server_rule_admin" model="ir.rule">
+        <field name="name">Mcp Server: all (admin)</field>
+        <field name="model_id" ref="model_mcp_server" />
+        <field name="domain_force">[(1, '=', 1)]</field>
+        <field name="groups" eval="[Command.link(ref('base.group_system'))]" />
+    </record>
+</odoo>

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")