fix: enforce record filter client-side (hosted scanner ignores the filter subobject)

sdk/python/aleo/_scanner_common.py

@@ -199,6 +199,25 @@ def build_owned_filter(
     return owned
 
 
+def enforce_record_filter(
+    records: list[OwnedRecord],
+    *,
+    program: str | None = None,
+    record: str | None = None,
+) -> list[OwnedRecord]:
+    """Apply the ``program``/``record`` subfilter to scanned records locally.
+
+    The hosted scanner ignores the ``filter`` subobject of an
+    :class:`OwnedFilter` and returns every record the account owns, so the
+    filter contract must be enforced on the results client-side.
+    """
+    if program is not None:
+        records = [r for r in records if r.get("program_name") == program]
+    if record is not None:
+        records = [r for r in records if r.get("record_name") == record]
+    return records
+
+
 def uuid_is_valid(uuid: str, network: str = "mainnet") -> bool:
     """Return True if uuid is a valid Field string (e.g. '1234...field')."""
     try:

sdk/python/aleo/facade/async_client.py

@@ -291,7 +291,11 @@ async def find(
             scanner.set_account(acct)
             scanner.set_decrypt_enabled(True)
 
-        from .._scanner_common import build_owned_filter, compute_uuid
+        from .._scanner_common import (
+            build_owned_filter,
+            compute_uuid,
+            enforce_record_filter,
+        )
 
         uuid = (
             str(compute_uuid(acct.view_key, self._client.provider.network))
@@ -303,8 +307,11 @@ async def find(
         )
 
         if amounts is not None:
-            return await scanner.find_credits_records(amounts, owned_filter)
-        return await scanner.find_records(owned_filter)
+            found = await scanner.find_credits_records(amounts, owned_filter)
+        else:
+            found = await scanner.find_records(owned_filter)
+        # The hosted scanner ignores the filter subobject; enforce it here.
+        return enforce_record_filter(found, program=program, record=record)
 
     async def find_credits(
         self, account: Any = None, at_least: int | None = None
@@ -336,7 +343,13 @@ async def find_credits(
         owned_filter = build_owned_filter(
             uuid, program="credits.aleo", record="credits"
         )
-        return await scanner.find_records(owned_filter)
+        from .._scanner_common import enforce_record_filter
+
+        return enforce_record_filter(
+            await scanner.find_records(owned_filter),
+            program="credits.aleo",
+            record="credits",
+        )
 
     async def get_unspent_credits_record(
         self,

sdk/python/aleo/facade/records.py

@@ -212,7 +212,11 @@ def find(
             scanner.set_account(acct)
             scanner.set_decrypt_enabled(True)
 
-        from .._scanner_common import build_owned_filter, compute_uuid
+        from .._scanner_common import (
+            build_owned_filter,
+            compute_uuid,
+            enforce_record_filter,
+        )
 
         uuid = (
             str(compute_uuid(acct.view_key, self._client._provider.network))
@@ -224,8 +228,11 @@ def find(
         )
 
         if amounts is not None:
-            return scanner.find_credits_records(amounts, owned_filter)
-        return scanner.find_records(owned_filter)
+            found = scanner.find_credits_records(amounts, owned_filter)
+        else:
+            found = scanner.find_records(owned_filter)
+        # The hosted scanner ignores the filter subobject; enforce it here.
+        return enforce_record_filter(found, program=program, record=record)
 
     def find_credits(
         self, account: Any = None, at_least: int | None = None
@@ -275,7 +282,13 @@ def find_credits(
         owned_filter = build_owned_filter(
             uuid, program="credits.aleo", record="credits"
         )
-        return scanner.find_records(owned_filter)
+        from .._scanner_common import enforce_record_filter
+
+        return enforce_record_filter(
+            scanner.find_records(owned_filter),
+            program="credits.aleo",
+            record="credits",
+        )
 
     # ── RecordProvider protocol ────────────────────────────────────────────────
 

sdk/python/tests/test_facade_async.py

@@ -279,6 +279,50 @@ async def fake_find_records(filt: Any) -> list[Any]:
     assert filt["filter"]["record"] == "credits"
 
 
+@pytest.mark.asyncio
+async def test_async_records_find_enforces_filter_client_side() -> None:
+    """find() filters by program/record locally — the hosted scanner ignores
+    the filter subobject and returns every record the account owns."""
+    a = AsyncAleo(HTTPProvider(BASE))
+    acct = _account(a)
+
+    mixed = [
+        {"program_name": "credits.aleo", "record_name": "credits", "spent": False},
+        {"program_name": "ethx_f466cc.aleo", "record_name": "Token", "spent": False},
+        {"program_name": "test_arc20_eth.aleo", "record_name": "Token", "spent": False},
+    ]
+
+    async def fake_find_records(filt: Any) -> list[Any]:
+        return mixed
+
+    a.records.scanner.set_account(acct)
+    a.records.scanner.find_records = fake_find_records  # type: ignore[method-assign]
+
+    records = await a.records.find(acct, program="test_arc20_eth.aleo")
+    assert [r["program_name"] for r in records] == ["test_arc20_eth.aleo"]
+
+
+@pytest.mark.asyncio
+async def test_async_records_find_credits_excludes_foreign_programs() -> None:
+    """find_credits() must not leak non-credits records the scanner returns."""
+    a = AsyncAleo(HTTPProvider(BASE))
+    acct = _account(a)
+
+    mixed = [
+        {"program_name": "credits.aleo", "record_name": "credits", "spent": False},
+        {"program_name": "test_arc20_eth.aleo", "record_name": "Token", "spent": False},
+    ]
+
+    async def fake_find_records(filt: Any) -> list[Any]:
+        return mixed
+
+    a.records.scanner.set_account(acct)
+    a.records.scanner.find_records = fake_find_records  # type: ignore[method-assign]
+
+    records = await a.records.find_credits(acct)
+    assert [r["program_name"] for r in records] == ["credits.aleo"]
+
+
 @pytest.mark.asyncio
 async def test_async_records_get_unspent_credits_record_covering() -> None:
     """get_unspent_credits_record returns a parsed RecordPlaintext when a covering record exists."""

sdk/python/tests/test_facade_records.py

@@ -225,6 +225,71 @@ def test_find_with_nonces() -> None:
     assert body["nonces"] == [RECORD_NONCE]
 
 
+MIXED_PROGRAM_RECORDS = [
+    {
+        "record_plaintext": RECORD_PLAINTEXT_STR,
+        "program_name": "credits.aleo",
+        "record_name": "credits",
+        "spent": False,
+    },
+    {
+        "record_plaintext": "{ owner: aleo1..., amount: 5u128.private }",
+        "program_name": "ethx_f466cc.aleo",
+        "record_name": "Token",
+        "spent": False,
+    },
+    {
+        "record_plaintext": "{ owner: aleo1..., amount: 7u128.private }",
+        "program_name": "test_arc20_eth.aleo",
+        "record_name": "Token",
+        "spent": False,
+    },
+]
+
+
+@resp_lib.activate
+def test_find_enforces_program_filter_client_side() -> None:
+    # The hosted scanner ignores the "filter" subobject and returns every
+    # record the account owns (verified live 2026-07-16); find() must enforce
+    # its documented program= contract on the results itself, or callers
+    # select records from the wrong program (phantom commitments on chain).
+    resp_lib.add(resp_lib.POST, f"{HOST}/records/owned", json=MIXED_PROGRAM_RECORDS)
+
+    a = _client()
+    _inject_scanner(a)
+    acct = _golden_account()
+
+    records = a.records.find(acct, program="test_arc20_eth.aleo")
+    assert [r["program_name"] for r in records] == ["test_arc20_eth.aleo"]
+
+
+@resp_lib.activate
+def test_find_enforces_record_name_filter_client_side() -> None:
+    resp_lib.add(resp_lib.POST, f"{HOST}/records/owned", json=MIXED_PROGRAM_RECORDS)
+
+    a = _client()
+    _inject_scanner(a)
+    acct = _golden_account()
+
+    records = a.records.find(acct, program="credits.aleo", record="credits")
+    assert [r["program_name"] for r in records] == ["credits.aleo"]
+    assert all(r["record_name"] == "credits" for r in records)
+
+
+@resp_lib.activate
+def test_find_credits_excludes_foreign_programs() -> None:
+    # Same service quirk as above, via the find_credits(at_least=None) path:
+    # non-credits records must not leak into the returned list.
+    resp_lib.add(resp_lib.POST, f"{HOST}/records/owned", json=MIXED_PROGRAM_RECORDS)
+
+    a = _client()
+    _inject_scanner(a)
+    acct = _golden_account()
+
+    records = a.records.find_credits(acct)
+    assert [r["program_name"] for r in records] == ["credits.aleo"]
+
+
 # ---------------------------------------------------------------------------
 # 4. find_credits — at_least filter
 # ---------------------------------------------------------------------------