security(studio): allowlist Studio variable-values endpoint

spp_studio_api_v2/services/variable_value_service.py

@@ -34,6 +34,22 @@ class VariableValueService:
     def __init__(self, env: Environment):
         self.env = env
 
+    def _data_api_pullable_accessors(self):
+        """cel_accessors whose cached values may be exposed through the API.
+
+        Mirrors the Data API allowlist (`spp.cel.variable._get_data_api_pullable_domain`):
+        only ordinary external-provider variables are returnable; DCI-backed
+        (inter-registry health/vital) and computed/scoring/aggregate values are
+        excluded so this endpoint cannot become an oracle for sensitive cached
+        data. Cache rows are keyed by the variable's cel_accessor.
+        """
+        if "spp.cel.variable" not in self.env:
+            return []
+        # sudo: the allowlist is system config; API authorization is the scope
+        # check at the router. The API client user has no spp.cel.variable ACL.
+        Variable = self.env["spp.cel.variable"].sudo()  # nosemgrep: odoo-sudo-without-context
+        return Variable.search(Variable._get_data_api_pullable_domain()).mapped("cel_accessor")
+
     def get_values_for_subject(
         self,
         partner_id: int,
@@ -66,11 +82,15 @@ def get_values_for_subject(
             return {}
         DataValue = self.env["spp.data.value"]
 
-        # Build domain
+        # Build domain. Restrict to API-pullable variables and the current
+        # company so this endpoint cannot disclose sensitive cached values
+        # (DCI/CRVS, scoring) or cross-company data.
         domain = [
             ("subject_model", "=", "res.partner"),
             ("subject_id", "=", partner_id),
             ("period_key", "=", period_key),
+            ("company_id", "=", self.env.company.id),
+            ("variable_name", "in", self._data_api_pullable_accessors()),
         ]
 
         # Filter by variable names if specified
@@ -136,11 +156,15 @@ def get_values_for_subjects(
             return {}
         DataValue = self.env["spp.data.value"]
 
-        # Build domain
+        # Build domain. Restrict to API-pullable variables and the current
+        # company (see get_values_for_subject) so sensitive cached values are
+        # not disclosed in bulk.
         domain = [
             ("subject_model", "=", "res.partner"),
             ("subject_id", "in", partner_ids),
             ("period_key", "=", period_key),
+            ("company_id", "=", self.env.company.id),
+            ("variable_name", "in", self._data_api_pullable_accessors()),
         ]
 
         if variable_names and "*" not in variable_names:

spp_studio_api_v2/tests/test_studio_router.py

@@ -62,32 +62,60 @@ def setUpClass(cls):
                 }
             )
 
+        # The Studio variable-values endpoint only exposes API-pullable
+        # (ordinary external-provider) variables, so the fixture variables are
+        # external-provider; otherwise their cached values are filtered out.
+        cls.value_provider = cls.env["spp.data.provider"].search([("code", "=", "studio_test_provider")], limit=1)
+        if not cls.value_provider:
+            cls.value_provider = cls.env["spp.data.provider"].create(
+                {"name": "Studio Test Provider", "code": "studio_test_provider"}
+            )
+
         # Get or create CEL variable (avoid unique constraint on cel_accessor)
         cls.cel_variable = cls.env["spp.cel.variable"].search(
             [("cel_accessor", "=", "age"), ("applies_to", "=", "individual")], limit=1
         )
+        _age_vals = {
+            "source_type": "external",
+            "external_provider_id": cls.value_provider.id,
+            "state": "active",
+            "category_id": cls.variable_category.id,
+        }
         if not cls.cel_variable:
             cls.cel_variable = cls.env["spp.cel.variable"].create(
                 {
                     "name": "Age",
                     "cel_accessor": "age",
-                    "source_type": "field",
                     "value_type": "number",
-                    "state": "active",
                     "applies_to": "individual",
-                    "source_model": "res.partner",
-                    "source_field": "age",
                     "period_granularity": "current",
                     "supports_historical": False,
-                    "category_id": cls.variable_category.id,
+                    **_age_vals,
                 }
             )
         else:
-            # Update existing variable to use the test category
-            cls.cel_variable.write(
+            # Update existing variable to the pullable (external-provider) shape.
+            cls.cel_variable.write(_age_vals)
+
+        # 'income' fixture used by the variable-filter tests (also pullable).
+        # spp_studio seeds non-external 'income' variables (standard_variables.xml),
+        # so make the existing ones pullable rather than skipping when present.
+        _income_vals = {
+            "source_type": "external",
+            "external_provider_id": cls.value_provider.id,
+            "state": "active",
+        }
+        income_vars = cls.env["spp.cel.variable"].search([("cel_accessor", "=", "income")])
+        if income_vars:
+            income_vars.write(_income_vals)
+        else:
+            cls.env["spp.cel.variable"].create(
                 {
-                    "category_id": cls.variable_category.id,
-                    "state": "active",
+                    "name": "Income",
+                    "cel_accessor": "income",
+                    "value_type": "number",
+                    "applies_to": "both",
+                    **_income_vals,
                 }
             )
 
@@ -845,8 +873,9 @@ def test_list_variables_returns_active_variables(self):
         self.assertIsNotNone(var_item)
         self.assertIn("Age", var_item["label"])  # May be "Age" or "Age (Years)"
         self.assertEqual(var_item["valueType"], "number")
-        # sourceType may be "field" or "computed" depending on demo data
-        self.assertIn(var_item["sourceType"], ["field", "computed"])
+        # sourceType may be "field"/"computed" (demo data) or "external" (the
+        # pullable fixture this suite uses for the subject-values endpoint).
+        self.assertIn(var_item["sourceType"], ["field", "computed", "external"])
         self.assertIn(var_item["appliesTo"], ["individual", "both"])
         self.assertIn(
             var_item["periodGranularity"],

spp_studio_api_v2/tests/test_variable_value_service.py

@@ -25,12 +25,18 @@ def setUp(self):
             }
         )
 
-        # Create test variable with unique name
+        # The Studio variable-values endpoint only exposes API-pullable
+        # (ordinary external-provider) variables, so the fixture variable is
+        # external-provider to exercise the "value is returned" path.
+        self.provider = self.env["spp.data.provider"].create(
+            {"name": f"Edu {self._test_id}", "code": f"edu_{self._test_id}"}
+        )
         self.test_variable = self.env["spp.cel.variable"].create(
             {
                 "name": self._var_name,
                 "cel_accessor": self._var_name,
-                "source_type": "field",
+                "source_type": "external",
+                "external_provider_id": self.provider.id,
                 "value_type": "number",
                 "state": "active",
                 "applies_to": "both",
@@ -134,6 +140,81 @@ def test_get_values_for_subjects_bulk(self):
         self.assertEqual(result[self.test_partner.id][self._var_name]["value"], 42)
         self.assertEqual(result[partner2.id][self._var_name]["value"], 100)
 
+    def _cache_value(self, variable_name, partner_id, value=7, source_type="external", company=None):
+        self.env["spp.data.value"].create(
+            {
+                "company_id": (company or self.env.company).id,
+                "variable_name": variable_name,
+                "subject_model": "res.partner",
+                "subject_id": partner_id,
+                "period_key": "current",
+                "value_json": {"value": value},
+                "value_type": "number",
+                "source_type": source_type,
+                "recorded_at": fields.Datetime.now(),
+            }
+        )
+
+    def test_excludes_non_pullable_variable_values(self):
+        """A non-external (e.g. scoring/computed) variable's cached value must
+        not be returned, even when '*'/None requests all values."""
+        scoring_accessor = f"{self._var_name}_score"
+        self.env["spp.cel.variable"].create(
+            {
+                "name": scoring_accessor,
+                "cel_accessor": scoring_accessor,
+                "source_type": "scoring",
+                "value_type": "number",
+                "state": "active",
+                "applies_to": "both",
+            }
+        )
+        self._cache_value(scoring_accessor, self.test_partner.id, value=99, source_type="scoring")
+
+        from ..services.variable_value_service import VariableValueService
+
+        result = VariableValueService(self.env).get_values_for_subject(
+            self.test_partner.id, variable_names=None, period_key="current"
+        )
+        self.assertIn(self._var_name, result)  # external-provider -> allowed
+        self.assertNotIn(scoring_accessor, result)  # scoring -> excluded
+
+        # Even an explicit request for the sensitive variable returns nothing.
+        explicit = VariableValueService(self.env).get_values_for_subject(
+            self.test_partner.id, variable_names=[scoring_accessor], period_key="current"
+        )
+        self.assertEqual(explicit, {})
+
+    def test_empty_allowlist_returns_nothing(self):
+        """When no variable is API-pullable, the endpoint returns nothing.
+
+        Locks in the fail-closed contract: an empty allowlist must produce
+        `("variable_name", "in", [])` (match nothing), never match-all.
+        """
+        from unittest.mock import patch
+
+        from ..services.variable_value_service import VariableValueService
+
+        service = VariableValueService(self.env)
+        with patch.object(VariableValueService, "_data_api_pullable_accessors", return_value=[]):
+            result = service.get_values_for_subject(
+                self.test_partner.id, variable_names=None, period_key="current"
+            )
+        self.assertEqual(result, {})
+
+    def test_company_isolation(self):
+        """Cached values from another company must not be returned."""
+        other = self.env["res.company"].create({"name": f"Other {self._test_id}"})
+        self._cache_value(self._var_name, self.test_partner.id, value=500, company=other)
+
+        from ..services.variable_value_service import VariableValueService
+
+        result = VariableValueService(self.env).get_values_for_subject(
+            self.test_partner.id, variable_names=[self._var_name], period_key="current"
+        )
+        # Only the current company's value (42 from setUp), never the other's 500.
+        self.assertEqual(result[self._var_name]["value"], 42)
+
     def test_get_available_variables(self):
         """Test listing available variables."""
         from ..services.variable_value_service import VariableValueService
@@ -328,6 +409,21 @@ def test_value_json_without_value_key(self):
             }
         )
 
+        # Back the cached value with a pullable (external-provider) variable so
+        # the endpoint returns it (the allowlist excludes non-external vars).
+        provider = self.env["spp.data.provider"].create({"name": f"P {unique_id}", "code": f"p_{unique_id}"})
+        self.env["spp.cel.variable"].create(
+            {
+                "name": var_name,
+                "cel_accessor": var_name,
+                "source_type": "external",
+                "external_provider_id": provider.id,
+                "value_type": "number",
+                "state": "active",
+                "applies_to": "both",
+            }
+        )
+
         # Create data value with non-standard value_json
         self.env["spp.data.value"].create(
             {
@@ -369,6 +465,20 @@ def test_value_json_complex_structure(self):
             }
         )
 
+        # Back the cached value with a pullable (external-provider) variable.
+        provider = self.env["spp.data.provider"].create({"name": f"P {unique_id}", "code": f"p_{unique_id}"})
+        self.env["spp.cel.variable"].create(
+            {
+                "name": var_name,
+                "cel_accessor": var_name,
+                "source_type": "external",
+                "external_provider_id": provider.id,
+                "value_type": "number",
+                "state": "active",
+                "applies_to": "both",
+            }
+        )
+
         # Create data value with complex value_json
         complex_value = {
             "value": {"nested": {"data": "here"}, "count": 5},