fix(nano): Stop sharing NanoRNG object between contracts

hathor/nanocontracts/rng.py

@@ -12,7 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Sequence, TypeVar
+from __future__ import annotations
+
+from typing import Any, Sequence, TypeVar
 
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms
 
@@ -21,11 +23,21 @@
 T = TypeVar('T')
 
 
-class NanoRNG:
+class NoMethodOverrideMeta(type):
+    __slots__ = ()
+
+    def __setattr__(cls, name: str, value: Any) -> None:
+        raise AttributeError(f'Cannot override method `{name}`')
+
+
+class NanoRNG(metaclass=NoMethodOverrideMeta):
     """Implement a deterministic random number generator that will be used by the sorter.
 
     This implementation uses the ChaCha20 encryption as RNG.
     """
+
+    __slots__ = ('__seed', '__encryptor', '__frozen')
+
     def __init__(self, seed: bytes) -> None:
         self.__seed = Hash(seed)
 
@@ -36,17 +48,42 @@ def __init__(self, seed: bytes) -> None:
         cipher = Cipher(algorithm, mode=None)
         self.__encryptor = cipher.encryptor()
 
+        self.__frozen = True
+
+    @classmethod
+    def create_with_shell(cls, seed: bytes) -> NanoRNG:
+        """Create a NanoRNG instance wrapped in a lightweight shell subclass.
+
+        This method dynamically creates a subclass of NanoRNG (a "shell" class) and instantiates it. The shell class is
+        useful to prevent sharing classes and objects among different contracts.
+        """
+        class ShellNanoRNG(NanoRNG):
+            __slots__ = ()
+
+        return ShellNanoRNG(seed=seed)
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        if getattr(self, '_NanoRNG__frozen', False):
+            raise AttributeError("Cannot assign methods to this object.")
+        super().__setattr__(name, value)
+
     @property
     def seed(self) -> Hash:
         """Return the seed used to create the RNG."""
         return self.__seed
 
+    def randbytes(self, size: int) -> bytes:
+        """Return a random string of bytes."""
+        assert size >= 1
+        ciphertext = self.__encryptor.update(b'\0' * size)
+        assert len(ciphertext) == size
+        return ciphertext
+
     def randbits(self, bits: int) -> int:
         """Return a random integer in the range [0, 2**bits)."""
-        # Generate 64-bit random string of bytes.
         assert bits >= 1
         size = (bits + 7) // 8
-        ciphertext = self.__encryptor.update(b'\0' * size)
+        ciphertext = self.randbytes(size)
         x = int.from_bytes(ciphertext, byteorder='little', signed=False)
         return x % (2**bits)
 

hathor/nanocontracts/runner/runner.py

@@ -148,6 +148,7 @@ def __init__(
         self._call_info: CallInfo | None = None
 
         self._rng: NanoRNG | None = NanoRNG(seed) if seed is not None else None
+        self._rng_per_contract: dict[ContractId, NanoRNG] = {}
 
         # Information about updated tokens in the current call via syscalls.
         self._updated_tokens_totals: defaultdict[TokenUid, int] = defaultdict(int)
@@ -762,7 +763,10 @@ def syscall_get_rng(self) -> NanoRNG:
         """Return the RNG for the current contract being executed."""
         if self._rng is None:
             raise ValueError('no seed was provided')
-        return self._rng
+        contract_id = self.get_current_contract_id()
+        if contract_id not in self._rng_per_contract:
+            self._rng_per_contract[contract_id] = NanoRNG.create_with_shell(seed=self._rng.randbytes(32))
+        return self._rng_per_contract[contract_id]
 
     def _internal_create_contract(self, contract_id: ContractId, blueprint_id: BlueprintId) -> None:
         """Create a new contract without calling the initialize() method."""

tests/nanocontracts/test_rng.py

@@ -5,6 +5,7 @@
 from hathor.nanocontracts.catalog import NCBlueprintCatalog
 from hathor.nanocontracts.exception import NCFail
 from hathor.nanocontracts.rng import NanoRNG
+from hathor.nanocontracts.types import ContractId
 from hathor.transaction import Transaction
 from tests.dag_builder.builder import TestDAGBuilder
 from tests.simulation.base import SimulatorTestCase
@@ -24,15 +25,30 @@ def nop(self, ctx: Context) -> None:
             raise NCFail('bad luck')
 
 
+class AttackerBlueprint(Blueprint):
+    target: ContractId
+
+    @public
+    def initialize(self, ctx: Context, target: ContractId) -> None:
+        self.target = target
+
+    @public
+    def attack(self, ctx: Context) -> None:
+        self.syscall.rng.random = lambda: 0.75  # type: ignore[method-assign]
+        self.syscall.call_public_method(self.target, 'nop', actions=[])
+
+
 class NCConsensusTestCase(SimulatorTestCase):
     __test__ = True
 
     def setUp(self):
         super().setUp()
 
         self.myblueprint_id = b'x' * 32
+        self.attacker_blueprint_id = b'y' * 32
         self.catalog = NCBlueprintCatalog({
-            self.myblueprint_id: MyBlueprint
+            self.myblueprint_id: MyBlueprint,
+            self.attacker_blueprint_id: AttackerBlueprint,
         })
 
         self.manager = self.simulator.create_peer()
@@ -51,6 +67,62 @@ def test_rng_consistency(self) -> None:
             v2 = [rng2.randbits(32) for _ in range(n)]
             assert v1 == v2
 
+    def test_rng_override(self) -> None:
+        seed = b'0' * 32
+        rng = NanoRNG(seed=seed)
+
+        with self.assertRaises(AttributeError, match='Cannot assign methods to this object.'):
+            rng.random = lambda self: 2  # type: ignore[method-assign, misc, assignment]
+
+        with self.assertRaises(AttributeError, match='Cannot assign methods to this object.'):
+            setattr(rng, 'random', lambda self: 2)
+
+        with self.assertRaises(AttributeError, match='Cannot assign methods to this object.'):
+            from types import MethodType
+            rng.random = MethodType(lambda self: 2, rng)  # type: ignore[method-assign]
+
+        with self.assertRaises(AttributeError, match='\'NanoRNG\' object attribute \'random\' is read-only'):
+            object.__setattr__(rng, 'random', lambda self: 2)
+
+        with self.assertRaises(AttributeError, match='AttributeError: Cannot override method `random`'):
+            NanoRNG.random = lambda self: 2  # type: ignore[method-assign]
+
+        with self.assertRaises(AttributeError, match='AttributeError: Cannot override method `random`'):
+            setattr(NanoRNG, 'random', lambda self: 2)
+
+        with self.assertRaises(TypeError, match='can\'t apply this __setattr__ to NoMethodOverrideMeta object'):
+            object.__setattr__(NanoRNG, 'random', lambda self: 2)
+
+        with self.assertRaises(AttributeError, match='AttributeError: Cannot override method `random`'):
+            rng.__class__.random = lambda self: 2  # type: ignore[method-assign]
+
+        with self.assertRaises(AttributeError, match='AttributeError: Cannot override method `random`'):
+            setattr(rng.__class__, 'random', lambda self: 2)
+
+        with self.assertRaises(TypeError, match='can\'t apply this __setattr__ to NoMethodOverrideMeta object'):
+            object.__setattr__(rng.__class__, 'random', lambda self: 2)
+
+        # mypy incorrectly infers the type of `rng.random` as `Never` (leading to "Never not callable [misc]")
+        # due to the override attempts above, which are expected to fail at runtime but confuse static analysis.
+        # This is a false positive in the test context; use `reveal_type(rng.random)` to inspect the inferred type.
+        assert rng.random() < 1  # type: ignore[misc]
+
+    def test_rng_shell_class(self) -> None:
+        seed = b'0' * 32
+        rng1 = NanoRNG.create_with_shell(seed=seed)
+        rng2 = NanoRNG.create_with_shell(seed=seed)
+
+        assert rng1.__class__ != rng2.__class__
+
+        with self.assertRaises(AttributeError, match='AttributeError: Cannot override method `random`'):
+            rng1.__class__.random = lambda self: 2  # type: ignore[method-assign]
+
+        with self.assertRaises(AttributeError, match='AttributeError: Cannot override method `random`'):
+            setattr(rng1.__class__, 'random', lambda self: 2)
+
+        with self.assertRaises(TypeError, match='can\'t apply this __setattr__ to NoMethodOverrideMeta object'):
+            object.__setattr__(rng1.__class__, 'random', lambda self: 2)
+
     def assertGoodnessOfFitTest(self, observed: list[int], expected: list[int]) -> None:
         """Pearson chi-square goodness-of-fit test for uniform [0, 1)"""
         assert len(observed) == len(expected)
@@ -259,3 +331,54 @@ def test_simple_rng(self) -> None:
         # For L = 3, it is 99.73%.
         # In other words, this assert should pass 99.73% of the runs.
         assert -L < z_score < L
+
+    def test_attack(self) -> None:
+        dag_builder = TestDAGBuilder.from_manager(self.manager)
+
+        n = 250
+        nc_calls_parts = []
+        for i in range(3, n + 3):
+            nc_calls_parts.append(f'''
+                nc{i}.nc_id = nc2
+                nc{i}.nc_method = attack()
+                nc{i} --> nc{i-1}
+            ''')
+        nc_calls = ''.join(nc_calls_parts)
+
+        artifacts = dag_builder.build_from_str(f'''
+            blockchain genesis b[1..33]
+            b30 < dummy
+
+            nc1.nc_id = "{self.myblueprint_id.hex()}"
+            nc1.nc_method = initialize()
+
+            nc2.nc_id = "{self.attacker_blueprint_id.hex()}"
+            nc2.nc_method = initialize(`nc1`)
+            nc2 --> nc1
+
+            {nc_calls}
+
+            nc{n+2} <-- b32
+        ''')
+
+        for node, vertex in artifacts.list:
+            assert self.manager.on_new_tx(vertex)
+
+        nc1, = artifacts.get_typed_vertices(['nc1'], Transaction)
+        assert nc1.is_nano_contract()
+        assert nc1.get_metadata().voided_by is None
+
+        names = [f'nc{i}' for i in range(3, n + 3)]
+        vertices = artifacts.get_typed_vertices(names, Transaction)
+
+        success = 0
+        fail = 0
+        for v in vertices:
+            assert v.is_nano_contract()
+            assert v.get_metadata().nc_execution is not None
+            if v.get_metadata().voided_by is None:
+                success += 1
+            else:
+                fail += 1
+        self.assertEqual(0, success)
+        self.assertEqual(n, fail)