Restrict DCI sync and cached value access

spp_cel_domain/security/ir.model.access.csv

@@ -1,5 +1,5 @@
 id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
-access_spp_data_value_user,spp.data.value user,model_spp_data_value,base.group_user,1,0,0,0
+access_spp_data_value_user,spp.data.value base user disabled,model_spp_data_value,base.group_user,0,0,0,0
 access_cel_rule_wizard_manager,access_cel_rule_wizard_manager,model_spp_cel_rule_wizard,spp_cel_domain.group_cel_domain_manager,1,1,1,0
 access_cel_rule_wizard_metric_manager,access_cel_rule_wizard_metric_manager,model_spp_cel_rule_wizard_metric,spp_cel_domain.group_cel_domain_manager,1,0,1,1
 access_cel_variable_category_viewer,access_cel_variable_category_viewer,model_spp_cel_variable_category,spp_cel_domain.group_cel_domain_viewer,1,0,0,0

spp_dci_indicators/data/dci_sync.xml

@@ -7,6 +7,10 @@
         <field name="binding_model_id" ref="base.model_res_partner" />
         <field name="binding_view_types">list,form</field>
         <field name="state">code</field>
+        <field
+            name="group_ids"
+            eval="[Command.link(ref('spp_cel_domain.group_cel_domain_manager'))]"
+        />
         <field name="code">
 if records:
     count = env["spp.dci.cel.fetcher"].sync_for_partners(records.ids)
@@ -28,6 +32,7 @@ if records:
         <field name="name">DCI: Sync CEL metrics</field>
         <field name="model_id" ref="base.model_res_partner" />
         <field name="state">code</field>
+        <field name="user_id" ref="base.user_root" />
         <field
             name="code"
         >env["spp.dci.cel.fetcher"].cron_sync_all_registrants()</field>

spp_dci_indicators/models/dci_cel_fetcher.py

@@ -13,7 +13,8 @@
 
 import logging
 
-from odoo import api, models
+from odoo import _, api, models
+from odoo.exceptions import AccessError
 
 _logger = logging.getLogger(__name__)
 
@@ -92,6 +93,14 @@ def _dci_backed_variables(self):
             ]
         )
 
+    @api.model
+    def _check_dci_sync_access(self):
+        """Require CEL manager privileges before triggering outbound DCI sync."""
+        if self.env.su:
+            return  # System/cron (e.g. the scheduled sync) runs in superuser mode
+        if not self.env.user.has_group("spp_cel_domain.group_cel_domain_manager"):
+            raise AccessError(_("Only CEL Domain Managers can sync DCI-backed CEL values."))
+
     @api.model
     def sync_for_partners(self, partner_ids, variables=None):
         """Fetch + cache all DCI-backed variables for the given partners.
@@ -100,6 +109,7 @@ def sync_for_partners(self, partner_ids, variables=None):
         manager's precompute path, which calls this fetcher and stores the result
         in spp.data.value.
         """
+        self._check_dci_sync_access()
         partner_ids = list(partner_ids or [])
         if not partner_ids:
             return 0

spp_dci_indicators/tests/test_dci_cel_fetcher.py

@@ -9,6 +9,8 @@
 
 from unittest.mock import patch
 
+from odoo import Command
+from odoo.exceptions import AccessError
 from odoo.tests import TransactionCase, tagged
 
 from odoo.addons.spp_dci.schemas.constants import RegistryType
@@ -151,6 +153,55 @@ def test_sync_for_partners_caches_values(self):
     def test_sync_for_partners_empty_is_noop(self):
         self.assertEqual(self.Fetcher.sync_for_partners([]), 0)
 
+    def test_sync_for_partners_requires_cel_manager(self):
+        plain_user = self.env["res.users"].create(
+            {
+                "name": "DCI Sync Plain User",
+                "login": "dci_sync_plain_user@example.test",
+                "group_ids": [Command.set([self.env.ref("base.group_user").id])],
+            }
+        )
+        with self.assertRaises(AccessError):
+            self.Fetcher.with_user(plain_user).sync_for_partners([self.partner.id], variables=self.var_is_alive)
+
+    def test_check_dci_sync_access_allows_cel_manager(self):
+        manager = self.env["res.users"].create(
+            {
+                "name": "DCI Sync CEL Manager",
+                "login": "dci_sync_cel_manager@example.test",
+                "group_ids": [
+                    Command.set(
+                        [
+                            self.env.ref("base.group_user").id,
+                            self.env.ref("spp_cel_domain.group_cel_domain_manager").id,
+                        ]
+                    )
+                ],
+            }
+        )
+        # A CEL Domain Manager must pass the access gate (no AccessError raised).
+        self.Fetcher.with_user(manager)._check_dci_sync_access()
+
+    def test_sync_for_partners_allows_superuser_for_cron(self):
+        """The scheduled cron runs in superuser mode (user_id=base.user_root).
+        A user without the CEL manager group must still sync when in su mode,
+        otherwise the cron would fail its own access check."""
+        plain_user = self.env["res.users"].create(
+            {
+                "name": "DCI Sync Cron User",
+                "login": "dci_sync_cron_user@example.test",
+                "group_ids": [Command.set([self.env.ref("base.group_user").id])],
+            }
+        )
+        self.assertFalse(plain_user.has_group("spp_cel_domain.group_cel_domain_manager"))
+        with patch(CHECK_DEATH, return_value=False):
+            count = (
+                self.Fetcher.with_user(plain_user)
+                .sudo()
+                .sync_for_partners([self.partner.id], variables=self.var_is_alive)
+            )
+        self.assertGreaterEqual(count, 1)
+
     def test_dci_backed_variables_excludes_plain_providers(self):
         plain = self.env["spp.data.provider"].create({"name": "Plain", "code": "plain_excl_t"})
         self.env["spp.cel.variable"].create(