add logic for ASSERT_MY_BIRTH_* conditions

chia/full_node/mempool_check_conditions.py

@@ -135,6 +135,12 @@ def mempool_check_time_locks(
 
     for spend in bundle_conds.spends:
         unspent = removal_coin_records[bytes32(spend.coin_id)]
+        if spend.birth_height is not None:
+            if spend.birth_height != unspent.confirmed_block_index:
+                return Err.ASSERT_MY_BIRTH_HEIGHT_FAILED
+        if spend.birth_seconds is not None:
+            if spend.birth_seconds != unspent.timestamp:
+                return Err.ASSERT_MY_BIRTH_SECONDS_FAILED
         if spend.height_relative is not None:
             if prev_transaction_block_height < unspent.confirmed_block_index + spend.height_relative:
                 return Err.ASSERT_HEIGHT_RELATIVE_FAILED

tests/blockchain/test_blockchain.py

@@ -1892,6 +1892,15 @@ async def test_aggsig_garbage(self, empty_blockchain, opcode, with_garbage, expe
     @pytest.mark.parametrize(
         "opcode,lock_value,expected,with_garbage",
         [
+            (co.ASSERT_MY_BIRTH_HEIGHT, -1, rbr.INVALID_BLOCK, False),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 0x100000000, rbr.INVALID_BLOCK, False),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 2, rbr.INVALID_BLOCK, False),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 3, rbr.NEW_PEAK, False),
+            (co.ASSERT_MY_BIRTH_SECONDS, -1, rbr.INVALID_BLOCK, False),
+            (co.ASSERT_MY_BIRTH_SECONDS, 0x10000000000000000, rbr.INVALID_BLOCK, False),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10029, rbr.INVALID_BLOCK, False),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10030, rbr.NEW_PEAK, False),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10031, rbr.INVALID_BLOCK, False),
             (co.ASSERT_SECONDS_RELATIVE, -2, rbr.NEW_PEAK, False),
             (co.ASSERT_SECONDS_RELATIVE, -1, rbr.NEW_PEAK, False),
             (co.ASSERT_SECONDS_RELATIVE, 0, rbr.NEW_PEAK, False),
@@ -1922,6 +1931,14 @@ async def test_ephemeral_timelock(self, opcode, lock_value, expected, with_garba
         else:
             constants = test_constants
 
+            # if the softfork is not active in this test, fixup all the
+            # tests to instead expect NEW_PEAK unconditionally
+            if opcode in [
+                ConditionOpcode.ASSERT_MY_BIRTH_HEIGHT,
+                ConditionOpcode.ASSERT_MY_BIRTH_SECONDS,
+            ]:
+                expected = ReceiveBlockResult.NEW_PEAK
+
         async with make_empty_blockchain(constants) as b:
 
             blocks = bt.get_consecutive_blocks(

tests/core/full_node/test_conditions.py

@@ -131,9 +131,19 @@ class TestConditions:
         "opcode,value,expected",
         [
             # the chain has 4 blocks, the spend is happening in the 5th block
-            # the coin being spent was created in the 3rd block
+            # the coin being spent was created in the 3rd block (i.e. block 2)
             # ensure invalid heights fail and pass correctly, depending on
             # which end of the range they exceed
+            (co.ASSERT_MY_BIRTH_HEIGHT, -1, Err.ASSERT_MY_BIRTH_HEIGHT_FAILED),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 0x100000000, Err.ASSERT_MY_BIRTH_HEIGHT_FAILED),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 3, Err.ASSERT_MY_BIRTH_HEIGHT_FAILED),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 2, None),
+            # genesis timestamp is 10000 and each block is 10 seconds
+            (co.ASSERT_MY_BIRTH_SECONDS, -1, Err.ASSERT_MY_BIRTH_SECONDS_FAILED),
+            (co.ASSERT_MY_BIRTH_SECONDS, 0x10000000000000000, Err.ASSERT_MY_BIRTH_SECONDS_FAILED),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10019, Err.ASSERT_MY_BIRTH_SECONDS_FAILED),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10020, None),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10021, Err.ASSERT_MY_BIRTH_SECONDS_FAILED),
             (co.ASSERT_HEIGHT_RELATIVE, -1, None),
             (co.ASSERT_HEIGHT_RELATIVE, 0, None),
             (co.ASSERT_HEIGHT_RELATIVE, 0x100000000, Err.ASSERT_HEIGHT_RELATIVE_FAILED),
@@ -160,6 +170,15 @@ class TestConditions:
     )
     async def test_condition(self, opcode, value, expected, bt, softfork2):
         conditions = Program.to(assemble(f"(({opcode[0]} {value}))"))
+
+        # when soft fork 2 is not active, these conditions are also not active,
+        # and never constrain the block
+        if not softfork2 and opcode in [
+            co.ASSERT_MY_BIRTH_HEIGHT,
+            co.ASSERT_MY_BIRTH_SECONDS,
+        ]:
+            expected = None
+
         await check_conditions(bt, conditions, expected_err=expected, softfork2=softfork2)
 
     @pytest.mark.asyncio

tests/core/mempool/test_mempool.py

@@ -371,6 +371,15 @@ async def test_basic_mempool_manager(self, two_nodes_one_block, wallet_a, self_h
     @pytest.mark.parametrize(
         "opcode,lock_value,expected",
         [
+            (co.ASSERT_MY_BIRTH_HEIGHT, -1, mis.FAILED),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 0x100000000, mis.FAILED),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 5, mis.FAILED),
+            (co.ASSERT_MY_BIRTH_HEIGHT, 6, mis.SUCCESS),
+            (co.ASSERT_MY_BIRTH_SECONDS, -1, mis.FAILED),
+            (co.ASSERT_MY_BIRTH_SECONDS, 0x10000000000000000, mis.FAILED),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10049, mis.FAILED),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10050, mis.SUCCESS),
+            (co.ASSERT_MY_BIRTH_SECONDS, 10051, mis.FAILED),
             (co.ASSERT_SECONDS_RELATIVE, -2, mis.SUCCESS),
             (co.ASSERT_SECONDS_RELATIVE, -1, mis.SUCCESS),
             (co.ASSERT_SECONDS_RELATIVE, 0, mis.SUCCESS),

tests/core/mempool/test_mempool_manager.py

@@ -99,6 +99,8 @@ async def instantiate_mempool_manager(
 
 def make_test_conds(
     *,
+    birth_height: Optional[uint32] = None,
+    birth_seconds: Optional[uint64] = None,
     height_relative: Optional[uint32] = None,
     height_absolute: uint32 = uint32(0),
     seconds_relative: uint64 = uint64(0),
@@ -113,8 +115,8 @@ def make_test_conds(
                 seconds_relative,
                 None,
                 None,
-                None,
-                None,
+                birth_height,
+                birth_seconds,
                 [],
                 [],
                 0,
@@ -223,6 +225,46 @@ def test_seconds_absolute(
             == expected_error
         )
 
+    @pytest.mark.parametrize(
+        "value,expected_error",
+        [
+            # the coin's birth height is
+            (9, Err.ASSERT_MY_BIRTH_HEIGHT_FAILED),
+            (10, None),
+            (11, Err.ASSERT_MY_BIRTH_HEIGHT_FAILED),
+        ],
+    )
+    def test_assert_my_birth_height(
+        self,
+        value: uint32,
+        expected_error: Optional[Err],
+    ) -> None:
+        conds = make_test_conds(birth_height=value)
+        assert (
+            mempool_check_time_locks(self.REMOVALS, conds, self.PREV_BLOCK_HEIGHT, self.PREV_BLOCK_TIMESTAMP)
+            == expected_error
+        )
+
+    @pytest.mark.parametrize(
+        "value,expected_error",
+        [
+            # the coin's birth timestamp is
+            (9999, Err.ASSERT_MY_BIRTH_SECONDS_FAILED),
+            (10000, None),
+            (10001, Err.ASSERT_MY_BIRTH_SECONDS_FAILED),
+        ],
+    )
+    def test_assert_my_birth_seconds(
+        self,
+        value: uint64,
+        expected_error: Optional[Err],
+    ) -> None:
+        conds = make_test_conds(birth_seconds=value)
+        assert (
+            mempool_check_time_locks(self.REMOVALS, conds, self.PREV_BLOCK_HEIGHT, self.PREV_BLOCK_TIMESTAMP)
+            == expected_error
+        )
+
 
 def test_compute_assert_height() -> None:
     c1 = Coin(bytes32(b"a" * 32), bytes32(b"b" * 32), 1337)