diff --git a/minichain/transaction.py b/minichain/transaction.py index 27f41f3..298056e 100644 --- a/minichain/transaction.py +++ b/minichain/transaction.py @@ -6,82 +6,70 @@ class Transaction: + _TX_FIELDS = frozenset({"sender", "receiver", "amount", "nonce", "data", "timestamp", "signature"}) + + def __setattr__(self, name, value) -> None: + if name in self._TX_FIELDS and getattr(self, "_sealed", False): + raise AttributeError(f"Transaction is sealed; cannot modify '{name}'") + super().__setattr__(name, value) + if name in self._TX_FIELDS and hasattr(self, "_cached_tx_id"): + super().__setattr__("_cached_tx_id", None) + + @staticmethod + def _normalize_ts(ts) -> int: + ts = int(ts) + return ts * 1000 if ts < 1e12 else ts + def __init__(self, sender, receiver, amount, nonce, data=None, signature=None, timestamp=None): - self.sender = sender # Public key (Hex str) - self.receiver = receiver # Public key (Hex str) or None for Deploy + self.sender = sender + self.receiver = receiver self.amount = amount self.nonce = nonce - self.data = data # Preserve None (do NOT normalize to "") - if timestamp is None: - self.timestamp = round(time.time() * 1000) # New tx: seconds → ms - elif timestamp > 1e12: - self.timestamp = int(timestamp) # Already in ms (from network) - else: - self.timestamp = round(timestamp * 1000) # Seconds → ms - self.signature = signature # Hex str + self.data = data + self.timestamp = self._normalize_ts(timestamp) if timestamp is not None else round(time.time() * 1000) + self.signature = signature + self._cached_tx_id = None + self._sealed = False def to_dict(self): - return { - "sender": self.sender, - "receiver": self.receiver, - "amount": self.amount, - "nonce": self.nonce, - "data": self.data, - "timestamp": self.timestamp, - "signature": self.signature, - } + return {"sender": self.sender, "receiver": self.receiver, "amount": self.amount, + "nonce": self.nonce, "data": self.data, "timestamp": self.timestamp, + "signature": self.signature} def to_signing_dict(self): - return { - "sender": self.sender, - "receiver": self.receiver, - "amount": self.amount, - "nonce": self.nonce, - "data": self.data, - "timestamp": self.timestamp, - } + return {"sender": self.sender, "receiver": self.receiver, "amount": self.amount, + "nonce": self.nonce, "data": self.data, "timestamp": self.timestamp} @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"), - ) + 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")) @property def hash_payload(self): - """Returns the bytes to be signed.""" return canonical_json_bytes(self.to_signing_dict()) @property def tx_id(self): - """Deterministic identifier for the signed transaction.""" - return canonical_json_hash(self.to_dict()) + if self._cached_tx_id is None: + self._cached_tx_id = canonical_json_hash(self.to_dict()) + return self._cached_tx_id def sign(self, signing_key: SigningKey): - # Validate that the signing key matches the sender if signing_key.verify_key.encode(encoder=HexEncoder).decode() != self.sender: raise ValueError("Signing key does not match sender") - signed = signing_key.sign(self.hash_payload) - self.signature = signed.signature.hex() + self.signature = signing_key.sign(self.hash_payload).signature.hex() + self._sealed = True def verify(self): if not self.signature: return False - try: - verify_key = VerifyKey(self.sender, encoder=HexEncoder) - verify_key.verify(self.hash_payload, bytes.fromhex(self.signature)) - return True - + VerifyKey(self.sender, encoder=HexEncoder).verify( + self.hash_payload, bytes.fromhex(self.signature)) except (BadSignatureError, CryptoError, ValueError, TypeError): - # Covers: - # - Invalid signature - # - Malformed public key hex - # - Invalid hex in signature return False + else: + return True \ No newline at end of file diff --git a/tests/test_cache.py b/tests/test_cache.py new file mode 100644 index 0000000..0e7265e --- /dev/null +++ b/tests/test_cache.py @@ -0,0 +1,78 @@ +import pytest +from unittest.mock import patch +from minichain.transaction import Transaction +from nacl.signing import SigningKey +from nacl.encoding import HexEncoder + +def test_tx_caching_efficiency(): + """ + Verifies that the expensive hashing math is only performed once + and skipped on subsequent accesses (Memoization proof). + """ + sk = SigningKey.generate() + sender_hex = sk.verify_key.encode(encoder=HexEncoder).decode() + tx = Transaction(sender=sender_hex, receiver="addr", amount=100, nonce=1) + + # We 'patch' the hashing function to count how many times it's called + with patch('minichain.transaction.canonical_json_hash') as mock_hash: + mock_hash.return_value = "mocked_hash_value" + + # 1. First Access: Should trigger the hash calculation + res1 = tx.tx_id + assert res1 == "mocked_hash_value" + assert mock_hash.call_count == 1 + + # 2. Second Access: Should return the cached value (count remains 1) + res2 = tx.tx_id + assert res2 == "mocked_hash_value" + assert mock_hash.call_count == 1 # <--- THIS proves the cache worked! + + # 3. Comprehensive Invalidation: Changing ANY field must clear the cache + mutations = { + "sender": "new_sender_hex", + "receiver": "new_receiver", + "amount": 200, + "nonce": 2, + "data": "new_data", + "timestamp": 1234567890, + "signature": "fake_signature_hex" + } + + expected_calls = 1 + for field, new_value in mutations.items(): + # Mutate the field dynamically + setattr(tx, field, new_value) + + # Prove the cache was instantly killed + assert tx._cached_tx_id is None, f"Cache failed to clear when mutating {field}" + + # Access ID again, which forces a re-calculation + _ = tx.tx_id + + # Prove the hashing math ran exactly one more time + expected_calls += 1 + assert mock_hash.call_count == expected_calls, f"Hash did not recalculate for {field}" + +def test_signed_tx_is_sealed(): + """Verifies that a signed transaction clears cache, changes ID, and cannot be modified.""" + sk = SigningKey.generate() + sender_hex = sk.verify_key.encode(encoder=HexEncoder).decode() + tx = Transaction(sender=sender_hex, receiver="bob", amount=100, nonce=1) + + # 1. Grab the ID before signing + unsigned_id = tx.tx_id + assert tx._cached_tx_id == unsigned_id + + # 2. Sign it + tx.sign(sk) + + # 3. Prove signing killed the old cache + assert tx._cached_tx_id is None + + # 4. Prove the new ID is totally different + signed_id = tx.tx_id + assert signed_id != unsigned_id + + # 5. Prove it's locked down (Sealed) + with pytest.raises(AttributeError, match="Transaction is sealed"): + tx.amount = 500 \ No newline at end of file