diff --git a/spp_cel_domain/security/ir.model.access.csv b/spp_cel_domain/security/ir.model.access.csv index 5718bfb45..313ac9ebd 100644 --- a/spp_cel_domain/security/ir.model.access.csv +++ b/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 diff --git a/spp_dci_indicators/data/dci_sync.xml b/spp_dci_indicators/data/dci_sync.xml index 8abd13479..2e9efa38e 100644 --- a/spp_dci_indicators/data/dci_sync.xml +++ b/spp_dci_indicators/data/dci_sync.xml @@ -7,6 +7,10 @@ list,form code + if records: count = env["spp.dci.cel.fetcher"].sync_for_partners(records.ids) @@ -28,6 +32,7 @@ if records: DCI: Sync CEL metrics code + env["spp.dci.cel.fetcher"].cron_sync_all_registrants() diff --git a/spp_dci_indicators/models/dci_cel_fetcher.py b/spp_dci_indicators/models/dci_cel_fetcher.py index d2d646719..8096b7d5b 100644 --- a/spp_dci_indicators/models/dci_cel_fetcher.py +++ b/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 diff --git a/spp_dci_indicators/tests/test_dci_cel_fetcher.py b/spp_dci_indicators/tests/test_dci_cel_fetcher.py index 422bdfa70..83799ef68 100644 --- a/spp_dci_indicators/tests/test_dci_cel_fetcher.py +++ b/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(