[IMP] cetmix_tower_server: Access rules

cetmix_tower_server/__manifest__.py

@@ -25,6 +25,7 @@
         "security/cx_tower_variable_value_security.xml",
         "security/cx_tower_plan_security.xml",
         "security/cx_tower_plan_line_security.xml",
+        "security/cx_tower_key_security.xml",
         "security/cx_tower_plan_line_action_security.xml",
         "security/cx_tower_plan_log_security.xml",
         "security/cx_tower_server_log_security.xml",

cetmix_tower_server/models/__init__.py

@@ -23,3 +23,4 @@
 from . import cx_tower_server_template
 from . import cetmix_tower
 from . import cx_tower_variable_option
+from . import res_partner

cetmix_tower_server/models/cx_tower_key.py

@@ -476,7 +476,7 @@ def _resolve_key_type_secret(self, reference, **kwargs):
             )
 
         # Fetch keys
-        keys = self.search(key_domain).sudo()
+        keys = self.sudo().search(key_domain)
         if not keys:
             return
 

cetmix_tower_server/models/res_partner.py

@@ -0,0 +1,16 @@
+# Copyright (C) 2025 Cetmix OÜ
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
+
+
+from odoo import fields, models
+
+
+class ResPartner(models.Model):
+    _inherit = "res.partner"
+
+    server_ids = fields.One2many(
+        string="Servers",
+        comodel_name="cx.tower.server",
+        inverse_name="partner_id",
+        help="Cetmix Tower servers that belong to this partner",
+    )

cetmix_tower_server/security/cx_tower_key_security.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<odoo>
+
+    <record id="cx_tower_key_rule_group_manager_access" model="ir.rule">
+        <field name="name">Tower Key: manager access rule</field>
+        <field name="model_id" ref="model_cx_tower_key" />
+        <field name="domain_force">
+            [
+            "|",
+            "&amp;",
+            "&amp;",
+            ("server_id", "=", False),
+            ("server_ssh_ids", "=", False),
+            ("partner_id", "=", False),
+            "|",
+            "|",
+            ("server_id.message_partner_ids", "in", [user.partner_id.id]),
+            ("server_ssh_ids.message_partner_ids", "in", [user.partner_id.id]),
+            ("partner_id.server_ids.message_partner_ids", "in", [user.partner_id.id]),
+            ]
+        </field>
+        <field name="groups" eval="[(4, ref('cetmix_tower_server.group_manager'))]" />
+    </record>
+
+    <record id="cx_tower_key_rule_group_root_access" model="ir.rule">
+        <field name="name">Tower Key: root access rule</field>
+        <field name="model_id" ref="model_cx_tower_key" />
+        <field name="domain_force">[(1, "=", 1)]</field>
+        <field name="groups" eval="[(4,ref('cetmix_tower_server.group_root'))]" />
+    </record>
+
+</odoo>

cetmix_tower_server/security/ir.model.access.csv

@@ -18,7 +18,6 @@ access_command_manager,Command->Manager,model_cx_tower_command,group_manager,1,1
 access_command_root,Command->Root,model_cx_tower_command,group_root,1,1,1,1
 access_execute_command_user,Execute Command->User,model_cx_tower_command_execute_wizard,group_user,1,1,1,1
 access_execute_plan_user,Execute Plan->User,model_cx_tower_plan_execute_wizard,group_user,1,1,1,1
-access_key_user,Key->User,model_cx_tower_key,group_user,1,0,0,0
 access_key_manager,Key->Manager,model_cx_tower_key,group_manager,1,1,1,0
 access_key_root,Key->Root,model_cx_tower_key,group_root,1,1,1,1
 access_command_log_user,Command Log->User,model_cx_tower_command_log,group_user,1,0,0,0

cetmix_tower_server/tests/test_command.py

@@ -511,6 +511,7 @@ def test_execute_command_with_keys(self):
 
     def test_user_access_rule(self):
         """Test user access rule"""
+
         # Create the test command
         test_command = self.Command.create({"name": "Test command"})
 
@@ -644,6 +645,49 @@ def test_user_access_rule(self):
             msg="Command name should be same",
         )
 
+    def test_user_access_rule_with_keys(self):
+        """Test user access rule"""
+
+        # create server key
+        server_key = self.Key.create(
+            {
+                "name": "server key",
+                "secret_value": "server key value",
+                "key_type": "s",
+                "server_id": self.server_test_1.id,
+            }
+        )
+
+        # Create the test command with server key
+        code = f"mkdir {server_key.reference_code}"
+        command_with_key = self.Command.create(
+            {"name": "Command with key", "code": code, "access_level": "1"}
+        )
+
+        # Remove bob from all cxtower_server groups
+        self.remove_from_group(
+            self.user_bob,
+            [
+                "cetmix_tower_server.group_user",
+                "cetmix_tower_server.group_manager",
+                "cetmix_tower_server.group_root",
+            ],
+        )
+
+        # Add user to group
+        self.add_to_group(self.user_bob, "cetmix_tower_server.group_user")
+        # User can execute command with key
+        self.server_test_1.with_user(self.user_bob).execute_command(
+            command_with_key,
+        )
+
+        # Add user to group_manager
+        self.add_to_group(self.user_bob, "cetmix_tower_server.group_manager")
+        # Manager can execute command with key
+        self.server_test_1.with_user(self.user_bob).execute_command(
+            command_with_key,
+        )
+
     def test_parse_ssh_command_result(self):
         """Test ssh command result parsing"""
 

cetmix_tower_server/tests/test_key.py

@@ -27,33 +27,24 @@ def test_key_creation(self):
             "Trailing and leading whitespaces must be removed from name",
         )
 
-    def test_key_access_rights(self):
-        """Test private key security features"""
+    def test_global_key_access_rights(self):
+        """Test global private key security features"""
 
-        # Default message returned instead of key value
-        SECRET_VALUE_PLACEHOLDER = self.Key.SECRET_VALUE_PLACEHOLDER
-
-        # Store key value
+        # Store global key value
         self.write_and_invalidate(self.key_1, **{"secret_value": "pepe"})
 
         # Get key value as Bob
         key_bob = self.key_1.with_user(self.user_bob)
 
         with self.assertRaises(AccessError):
-            key_value = key_bob.secret_value
+            key_value = key_bob.read(["secret_value"])
 
         # Add user to group
         self.add_to_group(self.user_bob, "cetmix_tower_server.group_user")
 
-        # Get value
-        key_value = key_bob.secret_value
-
-        # Ensure placeholder is used instead of the key value
-        self.assertEqual(
-            key_value,
-            SECRET_VALUE_PLACEHOLDER,
-            msg="Must return placeholder '{}'".format(SECRET_VALUE_PLACEHOLDER),
-        )
+        with self.assertRaises(AccessError):
+            # Get value
+            key_value = key_bob.read(["secret_value"])
 
         # Test write
         with self.assertRaises(AccessError):
@@ -72,10 +63,97 @@ def test_key_access_rights(self):
         with self.assertRaises(AccessError):
             key_bob.unlink()
 
-        # Add Bob to Root group and test write again
+        # Add Bob to Root group and test delete again
         self.add_to_group(self.user_bob, "cetmix_tower_server.group_root")
         key_bob.unlink()
 
+    def test_server_key_access_rights(self):
+        """Test server private key security features"""
+
+        # Store server key value
+        self.write_and_invalidate(
+            self.key_1, **{"secret_value": "pepe", "server_id": self.server_test_1.id}
+        )
+
+        # Get key value as Bob
+        key_bob = self.key_1.with_user(self.user_bob)
+
+        # Add user to group
+        self.add_to_group(self.user_bob, "cetmix_tower_server.group_user")
+
+        with self.assertRaises(AccessError):
+            # Get value
+            key_bob.read(["secret_value"])
+
+        # Test write
+        with self.assertRaises(AccessError):
+            self.write_and_invalidate(key_bob, **{"secret_value": "frog"})
+
+        # Add Bob as subscriber to server
+        self.server_test_1.message_subscribe([self.user_bob.partner_id.id])
+
+        # User cannot read key value
+        with self.assertRaises(AccessError):
+            key_bob.read(["secret_value"])
+
+        # Remove Bob from server followers
+        self.server_test_1.message_unsubscribe([self.user_bob.partner_id.id])
+
+        # Add Bob to Manager group
+        self.add_to_group(self.user_bob, "cetmix_tower_server.group_manager")
+
+        # Manager cannot read key value
+        with self.assertRaises(AccessError):
+            key_bob.read(["secret_value"])
+
+        # Add Bob as subscriber to server
+        self.server_test_1.message_subscribe([self.user_bob.partner_id.id])
+
+        # Now manager server subscriber can read and update this key
+        key_bob.read(["secret_value"])
+        key_bob.write({"secret_value": "dog"})
+
+    def test_partner_key_access_rights(self):
+        """Test partner private key security features"""
+        # create test partner
+        partner = self.env["res.partner"].create({"name": "test partner"})
+
+        # update server for this partner
+        self.write_and_invalidate(self.server_test_1, **{"partner_id": partner.id})
+
+        # Store key partner
+        self.write_and_invalidate(self.key_1, **{"partner_id": partner.id})
+
+        # Get key value as Bob
+        key_bob = self.key_1.with_user(self.user_bob)
+
+        # Add user to group
+        self.add_to_group(self.user_bob, "cetmix_tower_server.group_user")
+
+        # Ensure that user don't has access to key
+        with self.assertRaises(AccessError):
+            # Get value
+            key_bob.read(["secret_value"])
+
+        # Add Bob to Manager group
+        self.add_to_group(self.user_bob, "cetmix_tower_server.group_manager")
+
+        # Ensure that manager don't has access to key
+        with self.assertRaises(AccessError):
+            # Get value
+            key_bob.read(["secret_value"])
+
+        # Ensure that key has no servers
+        self.assertFalse(self.key_1.server_id)
+        self.assertFalse(self.key_1.server_ssh_ids)
+
+        # Add Bob as subscriber to server
+        partner.server_ids.message_subscribe([self.user_bob.partner_id.id])
+
+        # Now manager server subscriber can read and update this key
+        key_bob.read(["secret_value"])
+        key_bob.write({"secret_value": "dog"})
+
     def test_extract_key_strings(self):
         """Check if key strings are extracted properly"""
         code = (