diff --git a/spp_studio_api_v2/services/variable_value_service.py b/spp_studio_api_v2/services/variable_value_service.py index 50de90f5..8e616aab 100644 --- a/spp_studio_api_v2/services/variable_value_service.py +++ b/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: diff --git a/spp_studio_api_v2/tests/test_studio_router.py b/spp_studio_api_v2/tests/test_studio_router.py index 6d25ab78..b0df8972 100644 --- a/spp_studio_api_v2/tests/test_studio_router.py +++ b/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"], diff --git a/spp_studio_api_v2/tests/test_variable_value_service.py b/spp_studio_api_v2/tests/test_variable_value_service.py index b61374a0..c6d89f54 100644 --- a/spp_studio_api_v2/tests/test_variable_value_service.py +++ b/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},