refactor: Consolidate scattered validation logic

.gitignore

@@ -218,6 +218,9 @@ _minted*
 *.spell.bad
 *.spell.txt
 
+# MiniChain local persistence directories
+/.node*/
+
 # svg
 svg-inkscape/
 

main.py

@@ -27,7 +27,7 @@
 from nacl.encoding import HexEncoder
 
 from minichain import Transaction, Blockchain, Block, State, Mempool, P2PNetwork, mine_block
-from minichain.validators import is_valid_receiver
+
 
 
 logger = logging.getLogger(__name__)
@@ -67,7 +67,7 @@ def mine_and_process_block(chain, mempool, miner_pk):
         if tx.nonce < expected_nonce:
             stale_txs.append(tx)
             continue
-        if temp_state.validate_and_apply(tx):
+        if temp_state.apply_transaction(tx):
             mineable_txs.append(tx)
 
     if stale_txs:
@@ -212,7 +212,7 @@ async def cli_loop(sk, pk, chain, mempool, network):
                 print("  Usage: send <receiver_address> <amount>")
                 continue
             receiver = parts[1]
-            if not is_valid_receiver(receiver):
+            if not Transaction.is_valid_address(receiver):
                 print("  Invalid receiver format. Expected 40 or 64 hex characters.")
                 continue
             try:

minichain/block.py

@@ -101,19 +101,37 @@ def compute_hash(self) -> str:
 
     @classmethod
     def from_dict(cls, payload: dict):
-        transactions = [
-            Transaction.from_dict(tx_payload)
-            for tx_payload in payload.get("transactions", [])
-        ]
-        block = cls(
-            index=payload["index"],
-            previous_hash=payload["previous_hash"],
-            transactions=transactions,
-            timestamp=payload.get("timestamp"),
-            difficulty=payload.get("difficulty"),
-        )
-        block.nonce = payload.get("nonce", 0)
-        block.hash = payload.get("hash")
-        if "merkle_root" in payload:
-            block.merkle_root = payload["merkle_root"]
-        return block
+        try:
+            transactions = [
+                Transaction.from_dict(tx_payload)
+                for tx_payload in payload.get("transactions", [])
+            ]
+            if any(tx is None for tx in transactions):
+                return None
+
+            block = cls(
+                index=payload["index"],
+                previous_hash=payload["previous_hash"],
+                transactions=transactions,
+                timestamp=payload.get("timestamp"),
+                difficulty=payload.get("difficulty"),
+            )
+            block.nonce = payload.get("nonce", 0)
+            block.hash = payload.get("hash")
+            if "merkle_root" in payload:
+                # Accept payload value, but validate against computed value in is_valid().
+                block.merkle_root = payload["merkle_root"]
+            return block
+        except (KeyError, TypeError, ValueError):
+            return None
+
+    def is_valid(self):
+        if type(self.index) is not int or type(self.nonce) is not int or type(self.timestamp) is not int:
+            return False
+        if not isinstance(self.previous_hash, str) or (self.hash is not None and not isinstance(self.hash, str)):
+            return False
+        if self.merkle_root is not None and not isinstance(self.merkle_root, str):
+            return False
+        if self.merkle_root != _calculate_merkle_root(self.transactions):
+            return False
+        return all(tx.is_valid() for tx in self.transactions)

minichain/chain.py

@@ -71,7 +71,7 @@ def add_block(self, block):
             temp_state = self.state.copy()
 
             for tx in block.transactions:
-                result = temp_state.validate_and_apply(tx)
+                result = temp_state.apply_transaction(tx)
 
                 # Reject block if any transaction fails
                 if not result:

minichain/mempool.py

@@ -12,8 +12,8 @@ def __init__(self, max_size=1000, transactions_per_block=100):
         self.transactions_per_block = transactions_per_block
 
     def add_transaction(self, tx):
-        if not tx.verify():
-            logger.warning("Mempool: Invalid signature rejected")
+        if not tx.is_valid():
+            logger.warning("Mempool: Invalid transaction rejected")
             return False
 
         with self._lock:

minichain/p2p.py

@@ -10,7 +10,6 @@
 import logging
 
 from .serialization import canonical_json_hash
-from .validators import is_valid_receiver
 
 logger = logging.getLogger(__name__)
 
@@ -114,41 +113,13 @@ async def _handle_incoming(
         await self._notify_peer_connected(writer, "Network: Error during peer sync")
 
     def _validate_transaction_payload(self, payload):
+        from .transaction import Transaction
         if not isinstance(payload, dict):
             return False
-
-        required_fields = {
-            "sender": str,
-            "amount": int,
-            "nonce": int,
-            "timestamp": int,
-            "signature": str,
-        }
-        optional_fields = {
-            "receiver": (str, type(None)),
-            "data": (str, type(None)),
-        }
-        allowed_fields = set(required_fields) | set(optional_fields)
-
-        if set(payload) != allowed_fields:
-            return False
-
-        for field, expected_type in required_fields.items():
-            if not isinstance(payload.get(field), expected_type):
-                return False
-
-        for field, expected_type in optional_fields.items():
-            if not isinstance(payload.get(field), expected_type):
-                return False
-
-        if payload["amount"] <= 0:
-            return False
-
-        receiver = payload.get("receiver")
-        if receiver is not None and not is_valid_receiver(receiver):
+        tx = Transaction.from_dict(payload)
+        if not tx:
             return False
-
-        return True
+        return tx.is_valid()
 
     def _validate_sync_payload(self, payload):
         if not isinstance(payload, dict) or set(payload) != {"accounts"}:
@@ -176,36 +147,11 @@ def _validate_sync_payload(self, payload):
         return True
 
     def _validate_block_payload(self, payload):
+        from .block import Block
         if not isinstance(payload, dict):
             return False
-
-        required_fields = {
-            "index": int,
-            "previous_hash": str,
-            "merkle_root": (str, type(None)),
-            "transactions": list,
-            "timestamp": int,
-            "difficulty": (int, type(None)),
-            "nonce": int,
-            "hash": str,
-        }
-        optional_fields = {"miner": str}
-        allowed_fields = set(required_fields) | set(optional_fields)
-
-        if not set(payload).issubset(allowed_fields):
-            return False
-
-        for field, expected_type in required_fields.items():
-            if not isinstance(payload.get(field), expected_type):
-                return False
-
-        if "miner" in payload and not isinstance(payload["miner"], str):
-            return False
-
-        return all(
-            self._validate_transaction_payload(tx_payload)
-            for tx_payload in payload["transactions"]
-        )
+        block = Block.from_dict(payload)
+        return block is not None and block.is_valid()
 
     def _validate_message(self, message):
         if not isinstance(message, dict):

minichain/state.py

@@ -26,8 +26,8 @@ def get_account(self, address):
         return self.accounts[address]
 
     def verify_transaction_logic(self, tx):
-        if not tx.verify():
-            logger.error(f"Error: Invalid signature for tx from {tx.sender[:8]}...")
+        if not tx.is_valid():
+            logger.error("Error: Invalid transaction schema or signature")
             return False
 
         sender_acc = self.get_account(tx.sender)
@@ -50,18 +50,7 @@ def copy(self):
         new_state.contract_machine = ContractMachine(new_state) # Reinitialize contract_machine
         return new_state
 
-    def validate_and_apply(self, tx):
-        """
-        Validate and apply a transaction.
-        Returns the same success/failure shape as apply_transaction().
-        NOTE: Delegates to apply_transaction. Callers should use this for
-        semantic validation entry points.
-        """
-        # Semantic validation: amount must be an integer and non-negative
-        if not isinstance(tx.amount, int) or tx.amount < 0:
-            return False
-        # Further checks can be added here
-        return self.apply_transaction(tx)
+
 
     def apply_transaction(self, tx):
         """

minichain/transaction.py

@@ -1,4 +1,5 @@
 import time
+import re
 from nacl.signing import SigningKey, VerifyKey
 from nacl.encoding import HexEncoder
 from nacl.exceptions import BadSignatureError, CryptoError
@@ -41,17 +42,41 @@ def to_signing_dict(self):
             "timestamp": self.timestamp,
         }
 
+    @staticmethod
+    def is_valid_address(address):
+        return bool(re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", address))
+
     @classmethod
     def from_dict(cls, payload: dict):
-        return cls(
-            sender=payload["sender"],
-            receiver=payload.get("receiver"),
-            amount=payload["amount"],
-            nonce=payload["nonce"],
-            data=payload.get("data"),
-            signature=payload.get("signature"),
-            timestamp=payload.get("timestamp"),
-        )
+        try:
+            return cls(
+                sender=payload["sender"],
+                receiver=payload.get("receiver"),
+                amount=payload["amount"],
+                nonce=payload["nonce"],
+                data=payload.get("data"),
+                signature=payload.get("signature"),
+                timestamp=payload.get("timestamp"),
+            )
+        except (KeyError, TypeError):
+            return None
+
+    def is_valid(self):
+        """Unified, stateless validation (Types, Schema, Signatures)"""
+        if not isinstance(self.amount, int) or self.amount < 0:
+            return False
+        if not isinstance(self.nonce, int) or self.nonce < 0:
+            return False
+        if not isinstance(self.sender, str) or not self.is_valid_address(self.sender):
+            return False
+        if self.receiver is not None and (not isinstance(self.receiver, str) or not self.is_valid_address(self.receiver)):
+            return False
+        if self.data is not None and not isinstance(self.data, str):
+            return False
+        if not isinstance(self.timestamp, int) or self.timestamp <= 0:
+            return False
+
+        return self.verify()
 
     @property
     def hash_payload(self):

tests/test_protocol_hardening.py

@@ -105,7 +105,7 @@ async def test_block_schema_accepts_current_block_wire_format(self):
         sender_pk = sender_sk.verify_key.encode(encoder=HexEncoder).decode()
         receiver_pk = SigningKey.generate().verify_key.encode(encoder=HexEncoder).decode()
 
-        tx = Transaction(sender_pk, receiver_pk, 1, 0, timestamp=123)
+        tx = Transaction(sender_pk, receiver_pk, 1, 0, timestamp=1600000000000)
         tx.sign(sender_sk)
 
         block = Block(index=1, previous_hash="0" * 64, transactions=[tx], timestamp=456, difficulty=2)