From 16fdad54b4136963b1acb6622fc7102c97210626 Mon Sep 17 00:00:00 2001
From: Edwin Gonzales <edwin@openspp.org>
Date: Mon, 29 Jun 2026 20:11:49 +0800
Subject: [PATCH 1/2] security(dci): restrict Data API pull to non-sensitive
 provider variables
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The Data API GET /Data/pull required only the generic data:read scope, read
spp.data.value with sudo (bypassing the #234 ORM ACL), and matched any
variable_name with no provider/variable check (unlike push/invalidate). Once
eligibility precompute/refresh cached DCI-backed values, a data:read client
could pull sensitive CRVS facts (r.dci.crvs.is_alive, birth_verified) and
disability metrics for any subject. Scoring/PMT values were exposed the same
way, and /variables enumerated all of them.

Add a positive, default-deny allowlist on the resolved variable:
- spp.cel.variable.is_data_api_pullable() — true only for ordinary
  external-provider variables; computed/scoring/aggregate are not.
- spp_dci_indicators overrides it to also exclude DCI-backed providers
  (inter-registry data has its own consent/provider boundary).
- /pull resolves the requested name by cel_accessor (the cache key), denies
  with a uniform 403 if it resolves to nothing (fail-closed, covers orphan
  rows) or if any match is not pullable; also adds the missing company_id
  filter (cross-company read under sudo).
- /variables lists only pullable external-provider variables.

The internal ORM/RPC read by base.group_user was already closed by #234; this
covers the FastAPI sudo path it did not. Sibling oracles (Simulation metric()
expressions; subject-existence via sudo resolve) are tracked separately.

Tests: base rule (spp_cel_domain), DCI override (spp_dci_indicators), endpoint
behavior incl. unknown/non-pullable -> 403 and company isolation
(spp_api_v2_data). spp_api_v2_data 57/57, spp_cel_domain 620/620,
spp_dci_indicators 82/82.
---
 spp_api_v2_data/routers/data.py               | 27 +++++-
 spp_api_v2_data/tests/test_data_api.py        | 91 +++++++++++++++++--
 spp_cel_domain/models/cel_variable.py         | 15 +++
 spp_cel_domain/tests/__init__.py              |  1 +
 .../tests/test_data_api_pullable.py           | 40 ++++++++
 spp_dci_indicators/models/__init__.py         |  1 +
 spp_dci_indicators/models/cel_variable_dci.py | 21 +++++
 .../tests/test_data_provider_dci.py           | 34 +++++++
 8 files changed, 223 insertions(+), 7 deletions(-)
 create mode 100644 spp_cel_domain/tests/test_data_api_pullable.py
 create mode 100644 spp_dci_indicators/models/cel_variable_dci.py

diff --git a/spp_api_v2_data/routers/data.py b/spp_api_v2_data/routers/data.py
index d85a0f72a..f7d4b9d53 100644
--- a/spp_api_v2_data/routers/data.py
+++ b/spp_api_v2_data/routers/data.py
@@ -304,6 +304,21 @@ async def pull_values(
             detail="At least one subject_external_id is required",
         )
 
+    # Only ordinary external-provider variables may be pulled through this
+    # generic API. Resolve the requested name against the variable definitions
+    # by cel_accessor (the key DCI/cache rows are stored under); deny if it
+    # resolves to nothing (fail closed: orphan cache rows from a deleted or
+    # de-provisioned variable stay sensitive) or if ANY match is not pullable
+    # (e.g. DCI-backed inter-registry data or scoring/PMT values). A uniform
+    # 403 avoids signalling whether a denied name is sensitive or unknown.
+    Variable = env["spp.cel.variable"].sudo()  # nosemgrep: odoo-sudo-without-context
+    matched_variables = Variable.search([("cel_accessor", "=", variable)])
+    if not matched_variables or not all(v.is_data_api_pullable() for v in matched_variables):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Variable is not available through the data API",
+        )
+
     # Resolve to internal IDs
     subject_id_map = {}  # internal_id -> external_id
     for ext_id in external_ids:
@@ -317,6 +332,7 @@ async def pull_values(
     # Query cached values
     DataValue = env["spp.data.value"].sudo()  # nosemgrep: odoo-sudo-without-context
     domain = [
+        ("company_id", "=", env.company.id),
         ("variable_name", "=", variable),
         ("subject_id", "in", list(subject_id_map.keys())),
         ("period_key", "=", period_key or "current"),
@@ -425,7 +441,15 @@ async def list_variables(
         )
 
     Variable = env["spp.cel.variable"].sudo()  # nosemgrep: odoo-sudo-without-context
-    domain = [("active", "=", True)]
+    # Only ordinary external-provider variables are exchanged through the data
+    # API. Restrict to those so computed/scoring/aggregate variables are not
+    # enumerated; DCI-backed (inter-registry) variables are dropped below via
+    # is_data_api_pullable(), so their accessors are not disclosed either.
+    domain = [
+        ("active", "=", True),
+        ("source_type", "=", "external"),
+        ("external_provider_id", "!=", False),
+    ]
 
     if provider_code:
         Provider = env["spp.data.provider"].sudo()  # nosemgrep: odoo-sudo-without-context
@@ -457,6 +481,7 @@ async def list_variables(
             provider_code=v.external_provider_id.code if v.external_provider_id else None,
         )
         for v in variables
+        if v.is_data_api_pullable()
     ]
 
     return VariablesListResponse(total=total, items=items)
diff --git a/spp_api_v2_data/tests/test_data_api.py b/spp_api_v2_data/tests/test_data_api.py
index b76af69d0..a4b55db0b 100644
--- a/spp_api_v2_data/tests/test_data_api.py
+++ b/spp_api_v2_data/tests/test_data_api.py
@@ -741,13 +741,92 @@ async def test_pull_values_with_namespace_uri(self):
         assert len(result.items) == 1
 
     async def test_pull_values_no_results(self):
-        """Test GET /Data/pull with no matching values."""
+        """Test GET /Data/pull with a valid pullable variable but no cached rows."""
         from ..routers.data import pull_values
 
+        # school_attendance is an ordinary external-provider variable (pullable);
+        # no values were cached for it in this test, so the result is empty.
         result = await pull_values(
             env=self.env,
             api_client=self.api_client,
-            variable="nonexistent_variable",
+            variable="school_attendance",
+            subject_external_ids="EDU-001",
+            period_key="current",
+            _count=100,
+        )
+
+        assert result.total == 0
+        assert len(result.items) == 0
+
+    async def test_pull_values_rejects_unknown_variable(self):
+        """Unknown variable names are rejected (fail-closed), so orphaned cache
+        rows from a deleted/de-provisioned variable cannot be pulled."""
+        from ..routers.data import pull_values
+
+        with self.assertRaises(HTTPException) as cm:
+            await pull_values(
+                env=self.env,
+                api_client=self.api_client,
+                variable="nonexistent_variable",
+                subject_external_ids="EDU-001",
+                period_key="current",
+                _count=100,
+            )
+
+        assert cm.exception.status_code == status.HTTP_403_FORBIDDEN
+
+    async def test_pull_values_rejects_non_pullable_variable(self):
+        """A non-external variable (e.g. computed/scoring) must not be pullable
+        through the generic data API, even if its values are cached."""
+        from ..routers.data import pull_values
+
+        # computed_var has source_type='computed' -> not pullable. Even with a
+        # cached row present, the request is refused before any value is read.
+        self.env["spp.data.value"].create(
+            {
+                "variable_name": "computed_var",
+                "subject_id": self.partner1.id,
+                "period_key": "current",
+                "value_json": {"value": 1},
+                "value_type": "number",
+                "source_type": "computed",
+            }
+        )
+
+        with self.assertRaises(HTTPException) as cm:
+            await pull_values(
+                env=self.env,
+                api_client=self.api_client,
+                variable="computed_var",
+                subject_external_ids="EDU-001",
+                period_key="current",
+                _count=100,
+            )
+
+        assert cm.exception.status_code == status.HTTP_403_FORBIDDEN
+
+    async def test_pull_values_company_isolation(self):
+        """Cached values from another company must not be returned."""
+        from ..routers.data import pull_values
+
+        other_company = self.env["res.company"].create({"name": "Other Co"})
+        # A cached value for the same variable+subject but a different company.
+        self.env["spp.data.value"].create(
+            {
+                "company_id": other_company.id,
+                "variable_name": "school_attendance",
+                "subject_id": self.partner1.id,
+                "period_key": "current",
+                "value_json": {"value": 0.95},
+                "value_type": "number",
+                "source_type": "external",
+            }
+        )
+
+        result = await pull_values(
+            env=self.env,  # request runs in the default company
+            api_client=self.api_client,
+            variable="school_attendance",
             subject_external_ids="EDU-001",
             period_key="current",
             _count=100,
@@ -1167,13 +1246,13 @@ async def test_list_variables_success(self):
             _last_id=None,
         )
 
-        assert result.total >= 3
-        assert len(result.items) >= 3
-        # Check our test variables are in the list
+        # Only ordinary external-provider variables are exchanged via the data
+        # API; computed/scoring/aggregate variables are no longer enumerated.
         accessors = [v.cel_accessor for v in result.items]
         assert "school_attendance" in accessors
         assert "vaccination_status" in accessors
-        assert "computed_var" in accessors
+        assert "computed_var" not in accessors
+        assert all(v.source_type == "external" for v in result.items)
 
     async def test_list_variables_filter_by_provider(self):
         """Test GET /Data/variables?provider_code filters by provider."""
diff --git a/spp_cel_domain/models/cel_variable.py b/spp_cel_domain/models/cel_variable.py
index 0b7db9777..a6fa7c373 100644
--- a/spp_cel_domain/models/cel_variable.py
+++ b/spp_cel_domain/models/cel_variable.py
@@ -420,6 +420,21 @@ def get_cel_expression(self, program_id=None):
 
         return self.cel_accessor
 
+    def is_data_api_pullable(self):
+        """Whether this variable's cached values may be exposed through the
+        generic external Data API (``/Data/pull`` and ``/Data/variables``).
+
+        The Data API is the push/pull channel for ordinary external-provider
+        data, so only variables backed by an external provider are pullable.
+        Computed/scoring/aggregate variables (e.g. PMT scores) are internal and
+        must not be retrievable through this channel. The DCI indicators module
+        further excludes DCI-backed providers (inter-registry data has its own
+        consent/provider boundary) by overriding this method. Callers default to
+        deny when a requested name resolves to no variable (fail closed).
+        """
+        self.ensure_one()
+        return bool(self.source_type == "external" and self.external_provider_id)
+
     # ═══════════════════════════════════════════════════════════════════════
     # CONSTRAINTS
     # ═══════════════════════════════════════════════════════════════════════
diff --git a/spp_cel_domain/tests/__init__.py b/spp_cel_domain/tests/__init__.py
index 8efd78182..f9eb6bff3 100644
--- a/spp_cel_domain/tests/__init__.py
+++ b/spp_cel_domain/tests/__init__.py
@@ -9,6 +9,7 @@
 from . import test_cel_functions
 from . import test_cel_parser
 from . import test_cel_predicate_guard
+from . import test_data_api_pullable
 from . import test_cel_security
 from . import test_cel_service
 from . import test_cel_sql_generation
diff --git a/spp_cel_domain/tests/test_data_api_pullable.py b/spp_cel_domain/tests/test_data_api_pullable.py
new file mode 100644
index 000000000..7f6eacd36
--- /dev/null
+++ b/spp_cel_domain/tests/test_data_api_pullable.py
@@ -0,0 +1,40 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Tests for spp.cel.variable.is_data_api_pullable (base rule).
+
+Only ordinary external-provider variables may be exposed through the generic
+external Data API; computed/scoring/aggregate variables must not be.
+"""
+
+from odoo.tests import TransactionCase, tagged
+
+
+@tagged("post_install", "-at_install")
+class TestDataApiPullable(TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.provider = cls.env["spp.data.provider"].create({"name": "Edu", "code": "edu_pullable_t"})
+
+    def _var(self, accessor, source_type, provider=True):
+        return self.env["spp.cel.variable"].create(
+            {
+                "name": f"v_{accessor}",
+                "cel_accessor": accessor,
+                "source_type": source_type,
+                "external_provider_id": self.provider.id if provider else False,
+                "value_type": "number",
+            }
+        )
+
+    def test_external_with_provider_is_pullable(self):
+        self.assertTrue(self._var("pa_ext", "external").is_data_api_pullable())
+
+    def test_external_without_provider_is_not_pullable(self):
+        self.assertFalse(self._var("pa_ext_np", "external", provider=False).is_data_api_pullable())
+
+    def test_computed_is_not_pullable(self):
+        self.assertFalse(self._var("pa_computed", "computed", provider=False).is_data_api_pullable())
+
+    def test_scoring_is_not_pullable(self):
+        # Scoring values (e.g. PMT) land in the same cache but must not be pulled.
+        self.assertFalse(self._var("pa_scoring", "scoring", provider=False).is_data_api_pullable())
diff --git a/spp_dci_indicators/models/__init__.py b/spp_dci_indicators/models/__init__.py
index 708b32836..07f62f63d 100644
--- a/spp_dci_indicators/models/__init__.py
+++ b/spp_dci_indicators/models/__init__.py
@@ -4,3 +4,4 @@
 from . import dci_cel_fetcher
 from . import data_cache_manager_dci
 from . import cel_variable_resolver_dci
+from . import cel_variable_dci
diff --git a/spp_dci_indicators/models/cel_variable_dci.py b/spp_dci_indicators/models/cel_variable_dci.py
new file mode 100644
index 000000000..2ddd0198a
--- /dev/null
+++ b/spp_dci_indicators/models/cel_variable_dci.py
@@ -0,0 +1,21 @@
+"""Exclude DCI-backed variables from the generic external Data API.
+
+DCI-backed variables carry inter-registry data (disability, vital/civil status)
+fetched via the DCI protocol. That data has its own consent/provider boundary
+(the DCI server path) and must not be retrievable through the generic Data API
+push/pull channel, even once eligibility precompute has cached its values.
+"""
+
+from odoo import models
+
+
+class CelVariableDCI(models.Model):
+    _inherit = "spp.cel.variable"
+
+    def is_data_api_pullable(self):
+        # Ordinary external-provider variables remain pullable; DCI-backed ones
+        # are not, regardless of the base rule.
+        pullable = super().is_data_api_pullable()
+        if pullable and self.external_provider_id.is_dci_backed:
+            return False
+        return pullable
diff --git a/spp_dci_indicators/tests/test_data_provider_dci.py b/spp_dci_indicators/tests/test_data_provider_dci.py
index 8994114b2..980ba9b3b 100644
--- a/spp_dci_indicators/tests/test_data_provider_dci.py
+++ b/spp_dci_indicators/tests/test_data_provider_dci.py
@@ -56,3 +56,37 @@ def test_link_set_at_create(self):
         )
         self.assertTrue(provider.is_dci_backed)
         self.assertEqual(provider.dci_data_source_id, self.dci_source)
+
+    def test_dci_backed_variable_not_data_api_pullable(self):
+        """DCI-backed external variables must not be exposed via the generic
+        Data API, even though ordinary external-provider variables are."""
+        dci_provider = self.Provider.create(
+            {
+                "name": "DCI CRVS",
+                "code": "dci_crvs_pullable_t",
+                "dci_data_source_id": self.dci_source.id,
+            }
+        )
+        ordinary_provider = self.Provider.create({"name": "Edu", "code": "edu_pullable_t"})
+
+        dci_var = self.env["spp.cel.variable"].create(
+            {
+                "name": "dci.crvs.is_alive_t",
+                "cel_accessor": "r.dci.crvs.is_alive_t",
+                "source_type": "external",
+                "external_provider_id": dci_provider.id,
+                "value_type": "boolean",
+            }
+        )
+        ordinary_var = self.env["spp.cel.variable"].create(
+            {
+                "name": "school_attendance_t",
+                "cel_accessor": "school_attendance_t",
+                "source_type": "external",
+                "external_provider_id": ordinary_provider.id,
+                "value_type": "number",
+            }
+        )
+
+        self.assertFalse(dci_var.is_data_api_pullable())
+        self.assertTrue(ordinary_var.is_data_api_pullable())

From 5c31c51ba650e8ff43ba4f83e13f35de53a6b2f0 Mon Sep 17 00:00:00 2001
From: Edwin Gonzales <edwin@openspp.org>
Date: Mon, 29 Jun 2026 20:39:08 +0800
Subject: [PATCH 2/2] security(dci): address review on Data API pull/list
 hardening

Code review (Gemini) + Codecov follow-ups on the pull/list allowlist:

- list_variables: filter at the DB level via a new model method
  spp.cel.variable._get_data_api_pullable_domain() (overridden in
  spp_dci_indicators to also exclude DCI-backed providers), instead of a Python
  post-filter. This fixes the total count and cursor pagination, which were
  computed on the unfiltered domain.

- Real, executing endpoint tests: the existing Data API endpoint tests are
  `async def` on a plain TransactionCase, which unittest never awaits (silent
  no-ops) -- which is why the router changes showed as uncovered. Add
  test_data_api_pull_security.py driving the async endpoints via asyncio.run so
  they actually run: pull rejects unknown (fail-closed) and non-pullable
  variables with 403, enforces company isolation, returns ordinary values; list
  excludes non-external with a consistent total. Remove the 3 no-op async pull
  tests added earlier (superseded by these).

These running tests surfaced two pre-existing latent bugs in the endpoints
(both in functions touched here), now fixed:
- pull_values crashed (pydantic) on any cached value without a TTL, because an
  unset Odoo Datetime returns False, not None -> serialize expires_at as
  `rec.expires_at or None`.
- list_variables crashed (AttributeError) when spp_studio is not installed,
  since `description` is contributed by spp_studio -> use getattr.

Tests: spp_api_v2_data 59/59, spp_cel_domain 621/621, spp_dci_indicators 82/82.

NOTE: the repo-wide async-test-no-op problem (async def endpoint tests across
several spp_api_v2_* modules never execute) is pre-existing and tracked
separately; only this module's relevant tests are made real here.
---
 spp_api_v2_data/routers/data.py               |  24 ++-
 spp_api_v2_data/tests/__init__.py             |   1 +
 spp_api_v2_data/tests/test_data_api.py        |  77 --------
 .../tests/test_data_api_pull_security.py      | 167 ++++++++++++++++++
 spp_cel_domain/models/cel_variable.py         |  15 ++
 .../tests/test_data_api_pullable.py           |  11 ++
 spp_dci_indicators/models/cel_variable_dci.py |   8 +-
 .../tests/test_data_provider_dci.py           |   7 +
 8 files changed, 219 insertions(+), 91 deletions(-)
 create mode 100644 spp_api_v2_data/tests/test_data_api_pull_security.py

diff --git a/spp_api_v2_data/routers/data.py b/spp_api_v2_data/routers/data.py
index f7d4b9d53..b5ad4eab5 100644
--- a/spp_api_v2_data/routers/data.py
+++ b/spp_api_v2_data/routers/data.py
@@ -351,7 +351,9 @@ async def pull_values(
                 value=value,
                 period_key=rec.period_key,
                 recorded_at=rec.recorded_at,
-                expires_at=rec.expires_at,
+                # Odoo returns False (not None) for an unset Datetime; a value
+                # cached without a TTL must serialize as null, not crash.
+                expires_at=rec.expires_at or None,
                 is_stale=rec.is_stale,
                 source_type=rec.source_type,
             )
@@ -442,25 +444,21 @@ async def list_variables(
 
     Variable = env["spp.cel.variable"].sudo()  # nosemgrep: odoo-sudo-without-context
     # Only ordinary external-provider variables are exchanged through the data
-    # API. Restrict to those so computed/scoring/aggregate variables are not
-    # enumerated; DCI-backed (inter-registry) variables are dropped below via
-    # is_data_api_pullable(), so their accessors are not disclosed either.
-    domain = [
-        ("active", "=", True),
-        ("source_type", "=", "external"),
-        ("external_provider_id", "!=", False),
-    ]
+    # API. The pullable domain (extended by the DCI module to drop DCI-backed
+    # providers) filters at the DB level, so total/pagination stay consistent;
+    # computed/scoring/aggregate and DCI accessors are never enumerated.
+    domain = Variable._get_data_api_pullable_domain()
 
     if provider_code:
         Provider = env["spp.data.provider"].sudo()  # nosemgrep: odoo-sudo-without-context
         provider = Provider.search([("code", "=", provider_code)], limit=1)
         if provider:
-            domain.append(("external_provider_id", "=", provider.id))
+            domain = domain + [("external_provider_id", "=", provider.id)]
         else:
             return VariablesListResponse(total=0, items=[])
 
     if source_type:
-        domain.append(("source_type", "=", source_type))
+        domain = domain + [("source_type", "=", source_type)]
 
     # Apply cursor-based pagination
     if _last_id is not None:
@@ -473,7 +471,8 @@ async def list_variables(
         VariableInfo(
             name=v.name,
             cel_accessor=v.cel_accessor,
-            description=v.description or None,
+            # description is contributed by spp_studio; tolerate its absence.
+            description=getattr(v, "description", None) or None,
             value_type=v.value_type or "number",
             source_type=v.source_type or "computed",
             cache_strategy=v.cache_strategy or "none",
@@ -481,7 +480,6 @@ async def list_variables(
             provider_code=v.external_provider_id.code if v.external_provider_id else None,
         )
         for v in variables
-        if v.is_data_api_pullable()
     ]
 
     return VariablesListResponse(total=total, items=items)
diff --git a/spp_api_v2_data/tests/__init__.py b/spp_api_v2_data/tests/__init__.py
index 6945eb79e..e59da13d4 100644
--- a/spp_api_v2_data/tests/__init__.py
+++ b/spp_api_v2_data/tests/__init__.py
@@ -1,3 +1,4 @@
 # Part of OpenSPP. See LICENSE file for full copyright and licensing details.
 
 from . import test_data_api
+from . import test_data_api_pull_security
diff --git a/spp_api_v2_data/tests/test_data_api.py b/spp_api_v2_data/tests/test_data_api.py
index a4b55db0b..b82f9cdea 100644
--- a/spp_api_v2_data/tests/test_data_api.py
+++ b/spp_api_v2_data/tests/test_data_api.py
@@ -758,83 +758,6 @@ async def test_pull_values_no_results(self):
         assert result.total == 0
         assert len(result.items) == 0
 
-    async def test_pull_values_rejects_unknown_variable(self):
-        """Unknown variable names are rejected (fail-closed), so orphaned cache
-        rows from a deleted/de-provisioned variable cannot be pulled."""
-        from ..routers.data import pull_values
-
-        with self.assertRaises(HTTPException) as cm:
-            await pull_values(
-                env=self.env,
-                api_client=self.api_client,
-                variable="nonexistent_variable",
-                subject_external_ids="EDU-001",
-                period_key="current",
-                _count=100,
-            )
-
-        assert cm.exception.status_code == status.HTTP_403_FORBIDDEN
-
-    async def test_pull_values_rejects_non_pullable_variable(self):
-        """A non-external variable (e.g. computed/scoring) must not be pullable
-        through the generic data API, even if its values are cached."""
-        from ..routers.data import pull_values
-
-        # computed_var has source_type='computed' -> not pullable. Even with a
-        # cached row present, the request is refused before any value is read.
-        self.env["spp.data.value"].create(
-            {
-                "variable_name": "computed_var",
-                "subject_id": self.partner1.id,
-                "period_key": "current",
-                "value_json": {"value": 1},
-                "value_type": "number",
-                "source_type": "computed",
-            }
-        )
-
-        with self.assertRaises(HTTPException) as cm:
-            await pull_values(
-                env=self.env,
-                api_client=self.api_client,
-                variable="computed_var",
-                subject_external_ids="EDU-001",
-                period_key="current",
-                _count=100,
-            )
-
-        assert cm.exception.status_code == status.HTTP_403_FORBIDDEN
-
-    async def test_pull_values_company_isolation(self):
-        """Cached values from another company must not be returned."""
-        from ..routers.data import pull_values
-
-        other_company = self.env["res.company"].create({"name": "Other Co"})
-        # A cached value for the same variable+subject but a different company.
-        self.env["spp.data.value"].create(
-            {
-                "company_id": other_company.id,
-                "variable_name": "school_attendance",
-                "subject_id": self.partner1.id,
-                "period_key": "current",
-                "value_json": {"value": 0.95},
-                "value_type": "number",
-                "source_type": "external",
-            }
-        )
-
-        result = await pull_values(
-            env=self.env,  # request runs in the default company
-            api_client=self.api_client,
-            variable="school_attendance",
-            subject_external_ids="EDU-001",
-            period_key="current",
-            _count=100,
-        )
-
-        assert result.total == 0
-        assert len(result.items) == 0
-
     async def test_pull_values_subject_not_found(self):
         """Test GET /Data/pull with non-existent subject returns empty."""
         from ..routers.data import pull_values
diff --git a/spp_api_v2_data/tests/test_data_api_pull_security.py b/spp_api_v2_data/tests/test_data_api_pull_security.py
new file mode 100644
index 000000000..962c06790
--- /dev/null
+++ b/spp_api_v2_data/tests/test_data_api_pull_security.py
@@ -0,0 +1,167 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Security tests for the Data API pull/list endpoints.
+
+These drive the async router functions to completion via asyncio.run so the
+endpoint bodies actually execute (a bare ``async def test_*`` on a unittest
+TestCase is collected but never awaited, i.e. a silent no-op). They verify the
+allowlist guard added to /Data/pull and /Data/variables.
+"""
+
+import asyncio
+
+from odoo.tests.common import TransactionCase
+
+from fastapi import HTTPException, status
+
+
+class TestDataApiPullSecurity(TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+
+        org_type = cls.env.ref("spp_consent.org_type_government", raise_if_not_found=False)
+        if not org_type:
+            org_type = cls.env["spp.consent.org.type"].search([("code", "=", "government")], limit=1)
+
+        cls.partner = cls.env["res.partner"].create({"name": "Subject 1", "ref": "EDU-001"})
+
+        cls.provider = cls.env["spp.data.provider"].create(
+            {"name": "Edu Ministry", "code": "edu_sec_t", "active": True}
+        )
+        cls.ext_var = cls.env["spp.cel.variable"].create(
+            {
+                "name": "School Attendance Sec",
+                "cel_accessor": "school_attendance_sec",
+                "source_type": "external",
+                "external_provider_id": cls.provider.id,
+                "value_type": "number",
+                "cache_strategy": "ttl",
+            }
+        )
+        cls.computed_var = cls.env["spp.cel.variable"].create(
+            {
+                "name": "Computed Sec",
+                "cel_accessor": "computed_sec",
+                "source_type": "computed",
+                "value_type": "number",
+                "cache_strategy": "none",
+            }
+        )
+
+        api_partner = cls.env["res.partner"].create({"name": "API Partner Sec"})
+        cls.api_client = cls.env["spp.api.client"].create(
+            {
+                "name": "Sec Client",
+                "partner_id": api_partner.id,
+                "organization_type_id": org_type.id,
+            }
+        )
+        cls.env["spp.api.client.scope"].create({"client_id": cls.api_client.id, "resource": "data", "action": "read"})
+
+    def _run(self, coro):
+        return asyncio.run(coro)
+
+    def _cache(self, variable_name, source_type="external", company=None):
+        self.env["spp.data.value"].create(
+            {
+                "company_id": (company or self.env.company).id,
+                "variable_name": variable_name,
+                "subject_id": self.partner.id,
+                "period_key": "current",
+                "value_json": {"value": 0.95},
+                "value_type": "number",
+                "source_type": source_type,
+            }
+        )
+
+    # --- pull ---------------------------------------------------------------
+
+    def test_pull_ordinary_external_variable_succeeds(self):
+        from ..routers.data import pull_values
+
+        self._cache("school_attendance_sec")
+        result = self._run(
+            pull_values(
+                env=self.env,
+                api_client=self.api_client,
+                variable="school_attendance_sec",
+                subject_external_ids="EDU-001",
+                period_key="current",
+                _count=100,
+            )
+        )
+        self.assertEqual(result.total, 1)
+        self.assertEqual(result.items[0].value, 0.95)
+
+    def test_pull_unknown_variable_rejected(self):
+        from ..routers.data import pull_values
+
+        with self.assertRaises(HTTPException) as cm:
+            self._run(
+                pull_values(
+                    env=self.env,
+                    api_client=self.api_client,
+                    variable="does_not_exist",
+                    subject_external_ids="EDU-001",
+                    period_key="current",
+                    _count=100,
+                )
+            )
+        self.assertEqual(cm.exception.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_pull_non_pullable_variable_rejected(self):
+        from ..routers.data import pull_values
+
+        # computed_var has a cached row but is not pullable -> 403 before read.
+        self._cache("computed_sec", source_type="computed")
+        with self.assertRaises(HTTPException) as cm:
+            self._run(
+                pull_values(
+                    env=self.env,
+                    api_client=self.api_client,
+                    variable="computed_sec",
+                    subject_external_ids="EDU-001",
+                    period_key="current",
+                    _count=100,
+                )
+            )
+        self.assertEqual(cm.exception.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_pull_other_company_value_not_returned(self):
+        from ..routers.data import pull_values
+
+        other = self.env["res.company"].create({"name": "Other Co Sec"})
+        self._cache("school_attendance_sec", company=other)
+        result = self._run(
+            pull_values(
+                env=self.env,
+                api_client=self.api_client,
+                variable="school_attendance_sec",
+                subject_external_ids="EDU-001",
+                period_key="current",
+                _count=100,
+            )
+        )
+        self.assertEqual(result.total, 0)
+
+    # --- list ---------------------------------------------------------------
+
+    def test_list_variables_excludes_non_external(self):
+        from ..routers.data import list_variables
+
+        result = self._run(
+            list_variables(
+                env=self.env,
+                api_client=self.api_client,
+                provider_code=None,
+                source_type=None,
+                _count=500,
+                _last_id=None,
+            )
+        )
+        accessors = [v.cel_accessor for v in result.items]
+        self.assertIn("school_attendance_sec", accessors)
+        self.assertNotIn("computed_sec", accessors)
+        # total is computed on the same filtered domain, so it matches the items.
+        self.assertEqual(result.total, len(result.items))
+        self.assertTrue(all(v.source_type == "external" for v in result.items))
diff --git a/spp_cel_domain/models/cel_variable.py b/spp_cel_domain/models/cel_variable.py
index a6fa7c373..e9a3016c7 100644
--- a/spp_cel_domain/models/cel_variable.py
+++ b/spp_cel_domain/models/cel_variable.py
@@ -435,6 +435,21 @@ def is_data_api_pullable(self):
         self.ensure_one()
         return bool(self.source_type == "external" and self.external_provider_id)
 
+    @api.model
+    def _get_data_api_pullable_domain(self):
+        """Search domain selecting variables pullable through the Data API.
+
+        The DB-level counterpart of ``is_data_api_pullable`` (kept in sync with
+        it), so ``/Data/variables`` can filter, count and paginate correctly in
+        a single query. The DCI indicators module extends this to also exclude
+        DCI-backed providers.
+        """
+        return [
+            ("active", "=", True),
+            ("source_type", "=", "external"),
+            ("external_provider_id", "!=", False),
+        ]
+
     # ═══════════════════════════════════════════════════════════════════════
     # CONSTRAINTS
     # ═══════════════════════════════════════════════════════════════════════
diff --git a/spp_cel_domain/tests/test_data_api_pullable.py b/spp_cel_domain/tests/test_data_api_pullable.py
index 7f6eacd36..778b0fb91 100644
--- a/spp_cel_domain/tests/test_data_api_pullable.py
+++ b/spp_cel_domain/tests/test_data_api_pullable.py
@@ -38,3 +38,14 @@ def test_computed_is_not_pullable(self):
     def test_scoring_is_not_pullable(self):
         # Scoring values (e.g. PMT) land in the same cache but must not be pulled.
         self.assertFalse(self._var("pa_scoring", "scoring", provider=False).is_data_api_pullable())
+
+    def test_pullable_domain_selects_only_external_provider(self):
+        Variable = self.env["spp.cel.variable"]
+        ext = self._var("pa_dom_ext", "external")
+        computed = self._var("pa_dom_computed", "computed", provider=False)
+        scoring = self._var("pa_dom_scoring", "scoring", provider=False)
+
+        matched = Variable.search(Variable._get_data_api_pullable_domain())
+        self.assertIn(ext, matched)
+        self.assertNotIn(computed, matched)
+        self.assertNotIn(scoring, matched)
diff --git a/spp_dci_indicators/models/cel_variable_dci.py b/spp_dci_indicators/models/cel_variable_dci.py
index 2ddd0198a..b363f456a 100644
--- a/spp_dci_indicators/models/cel_variable_dci.py
+++ b/spp_dci_indicators/models/cel_variable_dci.py
@@ -6,7 +6,7 @@
 push/pull channel, even once eligibility precompute has cached its values.
 """
 
-from odoo import models
+from odoo import api, models
 
 
 class CelVariableDCI(models.Model):
@@ -19,3 +19,9 @@ def is_data_api_pullable(self):
         if pullable and self.external_provider_id.is_dci_backed:
             return False
         return pullable
+
+    @api.model
+    def _get_data_api_pullable_domain(self):
+        # Exclude DCI-backed providers at the DB level so /Data/variables counts
+        # and paginates over the same set is_data_api_pullable() would allow.
+        return super()._get_data_api_pullable_domain() + [("external_provider_id.is_dci_backed", "=", False)]
diff --git a/spp_dci_indicators/tests/test_data_provider_dci.py b/spp_dci_indicators/tests/test_data_provider_dci.py
index 980ba9b3b..0d225e857 100644
--- a/spp_dci_indicators/tests/test_data_provider_dci.py
+++ b/spp_dci_indicators/tests/test_data_provider_dci.py
@@ -90,3 +90,10 @@ def test_dci_backed_variable_not_data_api_pullable(self):
 
         self.assertFalse(dci_var.is_data_api_pullable())
         self.assertTrue(ordinary_var.is_data_api_pullable())
+
+        # The DB-level pullable domain must exclude DCI-backed variables too, so
+        # /Data/variables counts and paginates over the same set.
+        Variable = self.env["spp.cel.variable"]
+        matched = Variable.search(Variable._get_data_api_pullable_domain())
+        self.assertIn(ordinary_var, matched)
+        self.assertNotIn(dci_var, matched)