security(gis): restrict spatial stats to GIS-published indicators

spp_api_v2_gis/README.rst

@@ -156,6 +156,19 @@ Dependencies
 Changelog
 =========
 
+19.0.2.0.1
+~~~~~~~~~~
+
+- fix(security): restrict spatial/proximity statistics to GIS-published
+  indicators. Client-supplied ``variables`` were passed straight to the
+  aggregation service, which resolves names via ``sudo()`` against all
+  indicators and CEL variables, so a caller with only GIS/statistics
+  read could request unpublished indicator or raw CEL variable names and
+  receive their aggregates. Supplied variables are now filtered to the
+  same GIS publication allowlist (``spp.indicator.is_published_gis``)
+  already used for the default path; unpublished names are dropped
+  before aggregation.
+
 19.0.2.0.0
 ~~~~~~~~~~
 

spp_api_v2_gis/__manifest__.py

@@ -2,7 +2,7 @@
 {
     "name": "OpenSPP GIS API",
     "category": "OpenSPP/Integration",
-    "version": "19.0.2.0.0",
+    "version": "19.0.2.0.1",
     "sequence": 1,
     "author": "OpenSPP.org",
     "website": "https://github.com/OpenSPP/OpenSPP2",

spp_api_v2_gis/readme/HISTORY.md

@@ -1,3 +1,14 @@
+### 19.0.2.0.1
+
+- fix(security): restrict spatial/proximity statistics to GIS-published
+  indicators. Client-supplied ``variables`` were passed straight to the
+  aggregation service, which resolves names via ``sudo()`` against all
+  indicators and CEL variables, so a caller with only GIS/statistics read could
+  request unpublished indicator or raw CEL variable names and receive their
+  aggregates. Supplied variables are now filtered to the same GIS publication
+  allowlist (``spp.indicator.is_published_gis``) already used for the default
+  path; unpublished names are dropped before aggregation.
+
 ### 19.0.2.0.0
 
 - Initial migration to OpenSPP2

spp_api_v2_gis/services/spatial_query_service.py

@@ -354,14 +354,26 @@ def _compute_via_aggregation_service(self, registrant_ids, variables):
         # Create an explicit scope for the registrant IDs
         scope = build_explicit_scope(registrant_ids)
 
-        # Determine statistics to compute
-        statistics_to_compute = variables
-        if not statistics_to_compute:
-            # Use GIS-published statistics
-            # nosemgrep: odoo-sudo-without-context
-            Statistic = self.env["spp.indicator"].sudo()
-            gis_stats = Statistic.get_published_for_context("gis")
-            statistics_to_compute = [stat.name for stat in gis_stats] if gis_stats else None
+        # Enforce the GIS publication boundary as a positive allowlist: only
+        # indicators an admin published to GIS may be computed here. Client-
+        # supplied `variables` are otherwise resolved by the aggregation
+        # registry via sudo() against ALL indicators and CEL variables, which
+        # would let a caller with only gis/statistics read pull aggregates for
+        # names never published to GIS. Restrict both the supplied and default
+        # paths to the same published set; unknown/unpublished names are dropped
+        # (matching the existing "unknown variables -> valid response" contract
+        # and avoiding an existence oracle).
+        # nosemgrep: odoo-sudo-without-context
+        Statistic = self.env["spp.indicator"].sudo()
+        # Keep the recordset order (spp.indicator _order) for a deterministic
+        # default result; use the set only for the supplied-path membership test.
+        published_gis = Statistic.get_published_for_context("gis").mapped("name")
+        published_gis_names = set(published_gis)
+
+        if variables:
+            statistics_to_compute = [name for name in variables if name in published_gis_names]
+        else:
+            statistics_to_compute = published_gis
 
         if not statistics_to_compute:
             return {
@@ -400,6 +412,9 @@ def _convert_aggregation_result(self, agg_result, registrant_ids=None):
         result = {}
         grouped_stats = {}
 
+        # These names are already restricted to GIS-published indicators by
+        # _compute_via_aggregation_service, so this metadata lookup only ever
+        # sees published names (no is_published_gis filter needed here).
         # nosemgrep: odoo-sudo-without-context
         Statistic = self.env["spp.indicator"].sudo()
         statistic_by_name = {stat.name: stat for stat in Statistic.search([("name", "in", list(statistics.keys()))])}

spp_api_v2_gis/static/description/index.html

@@ -557,32 +557,47 @@ <h1>Dependencies</h1>
 <div class="contents local topic" id="contents">
 <ul class="simple">
 <li><a class="reference internal" href="#changelog" id="toc-entry-1">Changelog</a><ul>
-<li><a class="reference internal" href="#section-1" id="toc-entry-2">19.0.2.0.0</a></li>
+<li><a class="reference internal" href="#section-1" id="toc-entry-2">19.0.2.0.1</a></li>
+<li><a class="reference internal" href="#section-2" id="toc-entry-3">19.0.2.0.0</a></li>
 </ul>
 </li>
-<li><a class="reference internal" href="#bug-tracker" id="toc-entry-3">Bug Tracker</a></li>
-<li><a class="reference internal" href="#credits" id="toc-entry-4">Credits</a></li>
+<li><a class="reference internal" href="#bug-tracker" id="toc-entry-4">Bug Tracker</a></li>
+<li><a class="reference internal" href="#credits" id="toc-entry-5">Credits</a></li>
 </ul>
 </div>
 <div class="section" id="changelog">
 <h2><a class="toc-backref" href="#toc-entry-1">Changelog</a></h2>
 <div class="section" id="section-1">
-<h3><a class="toc-backref" href="#toc-entry-2">19.0.2.0.0</a></h3>
+<h3><a class="toc-backref" href="#toc-entry-2">19.0.2.0.1</a></h3>
+<ul class="simple">
+<li>fix(security): restrict spatial/proximity statistics to GIS-published
+indicators. Client-supplied <tt class="docutils literal">variables</tt> were passed straight to the
+aggregation service, which resolves names via <tt class="docutils literal">sudo()</tt> against all
+indicators and CEL variables, so a caller with only GIS/statistics
+read could request unpublished indicator or raw CEL variable names and
+receive their aggregates. Supplied variables are now filtered to the
+same GIS publication allowlist (<tt class="docutils literal">spp.indicator.is_published_gis</tt>)
+already used for the default path; unpublished names are dropped
+before aggregation.</li>
+</ul>
+</div>
+<div class="section" id="section-2">
+<h3><a class="toc-backref" href="#toc-entry-3">19.0.2.0.0</a></h3>
 <ul class="simple">
 <li>Initial migration to OpenSPP2</li>
 </ul>
 </div>
 </div>
 <div class="section" id="bug-tracker">
-<h2><a class="toc-backref" href="#toc-entry-3">Bug Tracker</a></h2>
+<h2><a class="toc-backref" href="#toc-entry-4">Bug Tracker</a></h2>
 <p>Bugs are tracked on <a class="reference external" href="https://github.com/OpenSPP/OpenSPP2/issues">GitHub Issues</a>.
 In case of trouble, please check there if your issue has already been reported.
 If you spotted it first, help us to smash it by providing a detailed and welcomed
 <a class="reference external" href="https://github.com/OpenSPP/OpenSPP2/issues/new?body=module:%20spp_api_v2_gis%0Aversion:%2019.0%0A%0A**Steps%20to%20reproduce**%0A-%20...%0A%0A**Current%20behavior**%0A%0A**Expected%20behavior**">feedback</a>.</p>
 <p>Do not contact contributors directly about support or help with technical issues.</p>
 </div>
 <div class="section" id="credits">
-<h2><a class="toc-backref" href="#toc-entry-4">Credits</a></h2>
+<h2><a class="toc-backref" href="#toc-entry-5">Credits</a></h2>
 </div>
 </div>
 <div class="section" id="authors">

spp_api_v2_gis/tests/__init__.py

@@ -9,4 +9,5 @@
 from . import test_spatial_query_service
 from . import test_statistics_endpoint
 from . import test_batch_query
+from . import test_gis_published_variables
 from . import test_proximity_query

spp_api_v2_gis/tests/test_gis_published_variables.py

@@ -0,0 +1,106 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Security: GIS spatial stats must honor the GIS publication boundary.
+
+`request.variables` is client-supplied and flows to the aggregation service,
+which resolves names via sudo() against all indicators and all CEL variables.
+Only names an admin published to GIS (`spp.indicator.is_published_gis`) may be
+computed — supplying an unpublished indicator name, or a raw CEL variable name,
+must not return an aggregate for it.
+"""
+
+from odoo.tests.common import TransactionCase, tagged
+
+
+@tagged("post_install", "-at_install")
+class TestGisPublishedVariables(TransactionCase):
+    """Supplied variables are restricted to the GIS-published allowlist."""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        Partner = cls.env["res.partner"]
+        Indicator = cls.env["spp.indicator"]
+        CelVariable = cls.env["spp.cel.variable"]
+
+        cls.group = Partner.create({"name": "GIS Pub Group", "is_registrant": True, "is_group": True})
+        cls.member = Partner.create({"name": "GIS Pub Member", "is_registrant": True, "is_group": False})
+        if "spp.group.membership" in cls.env:
+            cls.env["spp.group.membership"].create({"group": cls.group.id, "individual": cls.member.id})
+        cls.registrant_ids = [cls.group.id, cls.member.id]
+
+        def _make_var(name):
+            return CelVariable.create(
+                {
+                    "name": name,
+                    "cel_accessor": name,
+                    "source_type": "aggregate",
+                    "aggregate_type": "count",
+                    "aggregate_target": "members",
+                    "aggregate_filter": "true",
+                    "value_type": "number",
+                    "applies_to": "group",
+                    "state": "active",
+                }
+            )
+
+        # Published-to-GIS indicator (allowed).
+        cls.published_var = _make_var("gis_pub_var")
+        cls.published = Indicator.create(
+            {
+                "name": "gis_published_stat",
+                "label": "GIS Published Stat",
+                "variable_id": cls.published_var.id,
+                "format": "count",
+                "is_published_gis": True,
+            }
+        )
+        # Indicator NOT published to GIS (must be denied).
+        cls.unpublished_var = _make_var("gis_unpub_var")
+        cls.unpublished = Indicator.create(
+            {
+                "name": "gis_unpublished_stat",
+                "label": "GIS Unpublished Stat",
+                "variable_id": cls.unpublished_var.id,
+                "format": "count",
+                "is_published_gis": False,
+            }
+        )
+        # Raw CEL variable, never surfaced as a GIS indicator (must be denied).
+        cls.raw_var = _make_var("gis_raw_cel_var")
+
+    def _service(self):
+        from ..services.spatial_query_service import SpatialQueryService
+
+        return SpatialQueryService(self.env)
+
+    def _flat_stat_keys(self, result):
+        return {k for k in result.get("statistics", {}) if k != "_grouped"}
+
+    def test_unpublished_indicator_name_is_not_computed(self):
+        result = self._service()._compute_statistics(self.registrant_ids, ["gis_unpublished_stat"])
+        self.assertNotIn("gis_unpublished_stat", self._flat_stat_keys(result))
+
+    def test_raw_cel_variable_name_is_not_computed(self):
+        result = self._service()._compute_statistics(self.registrant_ids, ["gis_raw_cel_var"])
+        self.assertNotIn("gis_raw_cel_var", self._flat_stat_keys(result))
+
+    def test_published_indicator_name_is_computed(self):
+        result = self._service()._compute_statistics(self.registrant_ids, ["gis_published_stat"])
+        self.assertIn("gis_published_stat", self._flat_stat_keys(result))
+
+    def test_default_path_uses_published_only(self):
+        result = self._service()._compute_statistics(self.registrant_ids, [])
+        keys = self._flat_stat_keys(result)
+        self.assertIn("gis_published_stat", keys)
+        self.assertNotIn("gis_unpublished_stat", keys)
+
+    def test_mixed_list_keeps_published_drops_unpublished(self):
+        # Smuggling an unpublished name alongside a published one must not leak
+        # the unpublished aggregate; the published one is still returned.
+        result = self._service()._compute_statistics(
+            self.registrant_ids, ["gis_published_stat", "gis_unpublished_stat", "gis_raw_cel_var"]
+        )
+        keys = self._flat_stat_keys(result)
+        self.assertIn("gis_published_stat", keys)
+        self.assertNotIn("gis_unpublished_stat", keys)
+        self.assertNotIn("gis_raw_cel_var", keys)