From b6694b7c6fc3e3c199f18eb0aa9c11a656ab3a4b Mon Sep 17 00:00:00 2001 From: siddhant Date: Sun, 5 Jul 2026 21:44:13 +0530 Subject: [PATCH 1/2] remove code smells, duplicate codex and refactor --- main.py | 106 ++++++++++++++----------- minichain/block.py | 23 +++--- minichain/chain.py | 167 +++++++++++++++++---------------------- minichain/contract.py | 15 ++-- minichain/p2p.py | 71 ++++++++--------- minichain/persistence.py | 81 +++++++++---------- minichain/rpc.py | 9 +-- minichain/state.py | 48 +++++------ 8 files changed, 241 insertions(+), 279 deletions(-) diff --git a/main.py b/main.py index 35a46b0..4ed7870 100644 --- a/main.py +++ b/main.py @@ -18,6 +18,7 @@ import argparse import asyncio +import json import logging import re import sys @@ -47,6 +48,36 @@ def create_wallet(): return sk, pk +def request_chain(network, start_index, limit): + """Broadcast a chain_request for blocks starting at start_index.""" + req = {"type": "chain_request", "data": {"start_index": start_index, "limit": limit}} + asyncio.create_task(network._broadcast_raw(req)) + + +def parse_amount_fee(parts, start): + """Parse optional non-negative amount/fee at parts[start] and parts[start+1]. + Returns (amount, fee), or None if invalid (a message is printed for the user).""" + try: + amount = int(parts[start]) if len(parts) > start else 0 + fee = int(parts[start + 1]) if len(parts) > start + 1 else 0 + except ValueError: + print(" Amount and fee must be integers.") + return None + if amount < 0 or fee < 0: + print(" Amount and fee cannot be negative.") + return None + return amount, fee + + +async def submit_and_broadcast(mempool, network, tx, ok_msg, reject_msg): + """Add a signed tx to the mempool and broadcast it, printing the outcome.""" + if mempool.add_transaction(tx): + await network.broadcast_transaction(tx) + print(ok_msg) + else: + print(reject_msg) + + # ────────────────────────────────────────────── # Block mining # ────────────────────────────────────────────── @@ -117,7 +148,6 @@ def mine_and_process_block(chain, mempool, miner_pk): def make_network_handler(chain, mempool, network): """Return an async callback that processes incoming P2P messages.""" - from minichain.validators import ValidationStatus async def handler(data): msg_type = data.get("type") @@ -143,8 +173,7 @@ async def handler(data): peer_tip = payload.get("latest_block_index", 0) if peer_tip > chain.last_block.index: logger.info("📡 Peer %s is ahead (%d > %d). Initiating chunked sync...", peer_addr, peer_tip, chain.last_block.index) - req = {"type": "chain_request", "data": {"start_index": chain.last_block.index + 1, "limit": 500}} - asyncio.create_task(network._broadcast_raw(req)) + request_chain(network, chain.last_block.index + 1, 500) elif msg_type == "tx": try: @@ -178,13 +207,11 @@ async def handler(data): else: if block.index > chain.last_block.index + 1: logger.warning("📥 Received Block #%s — ahead of us (tip: %s). Requesting chunked sync...", block.index, chain.last_block.index) - req = {"type": "chain_request", "data": {"start_index": chain.last_block.index + 1, "limit": 500}} - asyncio.create_task(network._broadcast_raw(req)) + request_chain(network, chain.last_block.index + 1, 500) else: logger.warning("📥 Received Block #%s — rejected. Fork detected, trigger reorg sync.", block.index) # For a fork, request the full chain to use resolve_conflicts - req = {"type": "chain_request", "data": {"start_index": 0, "limit": 1000000}} # Request full chain for reorg - asyncio.create_task(network._broadcast_raw(req)) + request_chain(network, 0, 1000000) return status elif msg_type == "chain_request": @@ -234,17 +261,15 @@ async def handler(data): mempool.remove_transactions(block.transactions) else: logger.warning("❌ Sync failed at Block #%d. Fork detected. Requesting full chain.", block.index) - req = {"type": "chain_request", "data": {"start_index": 0, "limit": 1000000}} - asyncio.create_task(network._broadcast_raw(req)) + request_chain(network, 0, 1000000) all_added = False break - + # If we added all blocks and we hit the limit, request next batch if all_added and len(new_chain) == requested_limit: next_index = chain.last_block.index + 1 logger.info("📡 Requesting next batch from index %d", next_index) - req = {"type": "chain_request", "data": {"start_index": next_index, "limit": requested_limit}} - asyncio.create_task(network._broadcast_raw(req)) + request_chain(network, next_index, requested_limit) return handler @@ -360,11 +385,11 @@ async def cli_loop(sk, pk, chain, mempool, network, datadir: str | None = None): tx = Transaction(sender=pk, receiver=receiver, amount=amount, nonce=nonce, fee=fee, chain_id=chain.chain_id) tx.sign(sk) - if mempool.add_transaction(tx): - await network.broadcast_transaction(tx) - print(f" {C_GREEN}✅ Tx sent:{C_RESET} {amount} coins → {receiver[:12]}...") - else: - print(f" {C_RED}❌ Transaction rejected{C_RESET} (invalid sig, duplicate, or mempool full).") + await submit_and_broadcast( + mempool, network, tx, + f" {C_GREEN}✅ Tx sent:{C_RESET} {amount} coins → {receiver[:12]}...", + f" {C_RED}❌ Transaction rejected{C_RESET} (invalid sig, duplicate, or mempool full).", + ) # ── deploy ── elif cmd == "deploy": @@ -379,26 +404,20 @@ async def cli_loop(sk, pk, chain, mempool, network, datadir: str | None = None): print(f" File not found: {filepath}") continue - try: - amount = int(parts[2]) if len(parts) > 2 else 0 - fee = int(parts[3]) if len(parts) > 3 else 0 - except ValueError: - print(" Amount and fee must be integers.") - continue - - if amount < 0 or fee < 0: - print(" Amount and fee cannot be negative.") + parsed = parse_amount_fee(parts, 2) + if parsed is None: continue + amount, fee = parsed nonce = chain.state.get_account(pk).get("nonce", 0) tx = Transaction(sender=pk, receiver=None, amount=amount, nonce=nonce, fee=fee, data=code, chain_id=chain.chain_id) tx.sign(sk) - if mempool.add_transaction(tx): - await network.broadcast_transaction(tx) - print(f" ✅ Deploy Tx sent (nonce={nonce}). Mine a block to confirm.") - else: - print(" ❌ Deploy Transaction rejected.") + await submit_and_broadcast( + mempool, network, tx, + f" ✅ Deploy Tx sent (nonce={nonce}). Mine a block to confirm.", + " ❌ Deploy Transaction rejected.", + ) # ── call ── elif cmd == "call": @@ -410,27 +429,21 @@ async def cli_loop(sk, pk, chain, mempool, network, datadir: str | None = None): print(" Invalid receiver format. Expected 40 or 64 hex characters.") continue payload = parts[2] - - try: - amount = int(parts[3]) if len(parts) > 3 else 0 - fee = int(parts[4]) if len(parts) > 4 else 0 - except ValueError: - print(" Amount and fee must be integers.") - continue - if amount < 0 or fee < 0: - print(" Amount and fee cannot be negative.") + parsed = parse_amount_fee(parts, 3) + if parsed is None: continue + amount, fee = parsed nonce = chain.state.get_account(pk).get("nonce", 0) tx = Transaction(sender=pk, receiver=receiver, amount=amount, nonce=nonce, fee=fee, data=payload, chain_id=chain.chain_id) tx.sign(sk) - if mempool.add_transaction(tx): - await network.broadcast_transaction(tx) - print(f" ✅ Call Tx sent to {receiver[:12]}... (payload='{payload}'). Mine a block to confirm.") - else: - print(" ❌ Call Transaction rejected.") + await submit_and_broadcast( + mempool, network, tx, + f" ✅ Call Tx sent to {receiver[:12]}... (payload='{payload}'). Mine a block to confirm.", + " ❌ Call Transaction rejected.", + ) # ── mine ── elif cmd == "mine": @@ -545,8 +558,7 @@ async def run_node(port: int, host: str, connect_to: str | None, fund: int, data # When a new peer connects, send our hello so they can handshake async def on_peer_connected(writer): - import json as _json - sync_msg = _json.dumps({ + sync_msg = json.dumps({ "type": "hello", "data": { "chain_id": chain.chain_id, diff --git a/minichain/block.py b/minichain/block.py index 12bc141..96923e5 100644 --- a/minichain/block.py +++ b/minichain/block.py @@ -1,6 +1,6 @@ import time import hashlib -from typing import Optional # <-- Removed 'List' as requested +from typing import Optional from collections.abc import Sequence from .transaction import Transaction @@ -24,7 +24,6 @@ def _calculate_merkle_tree(hashes: Sequence[str]) -> Optional[str]: hashes_list = new_level return hashes_list[0] -# <-- Updated to Sequence to accept the frozen tuple def _calculate_merkle_root(transactions: Sequence[Transaction]) -> Optional[str]: if not transactions: return None @@ -65,9 +64,8 @@ def __init__( self.hash: Optional[str] = None self.state_root: Optional[str] = state_root self.receipt_root: Optional[str] = receipt_root - self.miner: Optional[str] = miner - # NEW: compute merkle roots once + # Compute merkle roots once self.merkle_root: Optional[str] = _calculate_merkle_root(self.transactions) # If receipt_root is missing but we have receipts, calculate it. @@ -88,7 +86,7 @@ def to_header_dict(self): "difficulty": self.difficulty, "nonce": self.nonce, } - # Include miner in header only when present (optional field) <-- Reworded comment + # Include miner in header only when present (optional field) if self.miner is not None: header["miner"] = self.miner return header @@ -132,27 +130,26 @@ def from_dict(cls, payload: dict): for r_payload in payload.get("receipts", []) ] - # Safely extract and cast difficulty if it exists + # Safely extract and cast difficulty and timestamp if they exist raw_diff = payload.get("difficulty") parsed_diff = int(raw_diff) if raw_diff is not None else None - - # Safely extract and cast timestamp if it exists <-- Added explicit timestamp casting + raw_ts = payload.get("timestamp") parsed_ts = int(raw_ts) if raw_ts is not None else None block = cls( - index=int(payload["index"]), + index=int(payload["index"]), previous_hash=payload["previous_hash"], transactions=transactions, - timestamp=parsed_ts, # <-- Passed the casted timestamp - difficulty=parsed_diff, + timestamp=parsed_ts, + difficulty=parsed_diff, state_root=payload.get("state_root"), receipt_root=payload.get("receipt_root"), receipts=receipts, miner=payload.get("miner"), ) - block.nonce = int(payload.get("nonce", 0)) + block.nonce = int(payload.get("nonce", 0)) block.hash = payload.get("hash") - + # Verify the block hash expected_hash = block.compute_hash() if block.hash is not None and block.hash != expected_hash: diff --git a/minichain/chain.py b/minichain/chain.py index 8aacd17..fa7d0dd 100644 --- a/minichain/chain.py +++ b/minichain/chain.py @@ -120,6 +120,63 @@ def get_total_work(self, chain_list=None): chain_list = self.chain return sum(2 ** (block.difficulty or 1) for block in chain_list) + def _next_difficulty(self, difficulty, avg_block_time): + """Advance the EMA difficulty control after a block, returning the new value.""" + if avg_block_time > self.target_block_time: + return max(1, difficulty - 1) + if avg_block_time < self.target_block_time: + return difficulty + 1 + return difficulty + + def _apply_block(self, prev_block, block, state, difficulty, avg_block_time): + """ + Canonical block-application pipeline shared by add_block and resolve_conflicts. + Validates `block` against `prev_block` and applies its transactions to `state` + (mutated in place). On any non-VALID status the caller must discard `state`. + Returns: (ValidationStatus, new_difficulty, new_avg_block_time) + """ + from .validators import ValidationStatus + + try: + validate_block_link_and_hash(prev_block, block) + except ValueError as exc: + logger.warning("Block %s rejected: %s", block.index, exc) + status = ValidationStatus.INVALID if "hash" in str(exc) else ValidationStatus.FAILED + return status, difficulty, avg_block_time + + if block.difficulty != difficulty: + logger.warning("Block %s rejected: Invalid difficulty. Expected %s, got %s", block.index, difficulty, block.difficulty) + return ValidationStatus.INVALID, difficulty, avg_block_time + + receipts = [] + for tx in block.transactions: + status, receipt = state.validate_and_apply_with_status(tx) + if status != ValidationStatus.VALID: + logger.warning("Block %s rejected: Transaction failed validation", block.index) + return status, difficulty, avg_block_time + receipts.append(receipt) + + total_fees = sum(getattr(r, 'gas_used', 0) for r in receipts) + if block.miner: + state.credit_mining_reward(block.miner, reward=state.DEFAULT_MINING_REWARD + total_fees) + + computed_receipt_root = calculate_receipt_root(receipts) + if block.receipt_root != computed_receipt_root: + logger.warning("Block %s rejected: Invalid receipt root. Expected %s, got %s", block.index, computed_receipt_root, block.receipt_root) + return ValidationStatus.INVALID, difficulty, avg_block_time + + if [r.to_dict() for r in block.receipts] != [r.to_dict() for r in receipts]: + logger.warning("Block %s rejected: Receipts payload mismatch", block.index) + return ValidationStatus.INVALID, difficulty, avg_block_time + + computed_state_root = state.state_root() + if block.state_root != computed_state_root: + logger.warning("Block %s rejected: Invalid state root. Expected %s, got %s", block.index, computed_state_root, block.state_root) + return ValidationStatus.INVALID, difficulty, avg_block_time + + new_avg = self.alpha * (block.timestamp - prev_block.timestamp) + (1 - self.alpha) * avg_block_time + return ValidationStatus.VALID, self._next_difficulty(difficulty, new_avg), new_avg + def add_block(self, block): """ Validates and adds a block to the chain if all transactions succeed. @@ -128,60 +185,18 @@ def add_block(self, block): from .validators import ValidationStatus with self._lock: - try: - validate_block_link_and_hash(self.last_block, block) - except ValueError as exc: - logger.warning("Block %s rejected: %s", block.index, exc) - return ValidationStatus.INVALID if "hash" in str(exc) else ValidationStatus.FAILED - - if block.difficulty != self.current_difficulty: - logger.warning("Block %s rejected: Invalid difficulty. Expected %s, got %s", block.index, self.current_difficulty, block.difficulty) - return ValidationStatus.INVALID - - # Validate transactions on a temporary state copy temp_state = self.state.copy() temp_state.chain_id = self.chain_id - receipts = [] - - for tx in block.transactions: - status, receipt = temp_state.validate_and_apply_with_status(tx) - - # Reject block if any transaction fails mathematical validation - if status != ValidationStatus.VALID: - logger.warning("Block %s rejected: Transaction failed validation", block.index) - return status - - receipts.append(receipt) - - total_fees = sum(getattr(r, 'gas_used', 0) for r in receipts) - if block.miner: - temp_state.credit_mining_reward(block.miner, reward=temp_state.DEFAULT_MINING_REWARD + total_fees) - - computed_receipt_root = calculate_receipt_root(receipts) - if block.receipt_root != computed_receipt_root: - logger.warning("Block %s rejected: Invalid receipt root. Expected %s, got %s", block.index, computed_receipt_root, block.receipt_root) - return ValidationStatus.INVALID - - if [r.to_dict() for r in block.receipts] != [r.to_dict() for r in receipts]: - logger.warning("Block %s rejected: Receipts payload mismatch", block.index) - return ValidationStatus.INVALID - - # Verify state root - if block.state_root != temp_state.state_root(): - logger.warning("Block %s rejected: Invalid state root. Expected %s, got %s", block.index, temp_state.state_root(), block.state_root) - return ValidationStatus.INVALID - - # Update EMA difficulty state - time_diff = block.timestamp - self.last_block.timestamp - self.avg_block_time = self.alpha * time_diff + (1 - self.alpha) * self.avg_block_time - - if self.avg_block_time > self.target_block_time: - self.current_difficulty = max(1, self.current_difficulty - 1) - elif self.avg_block_time < self.target_block_time: - self.current_difficulty += 1 + status, new_difficulty, new_avg = self._apply_block( + self.last_block, block, temp_state, self.current_difficulty, self.avg_block_time + ) + if status != ValidationStatus.VALID: + return status # All transactions valid → commit state and append block self.state = temp_state + self.current_difficulty = new_difficulty + self.avg_block_time = new_avg self.chain.append(block) return ValidationStatus.VALID @@ -191,6 +206,8 @@ def resolve_conflicts(self, new_chain_list) -> tuple[bool, list]: attempts a reorg. Rebuilds state from genesis to guarantee validity. Returns: (success_bool, list_of_orphaned_transactions) """ + from .validators import ValidationStatus + if not new_chain_list: return False, [] @@ -220,55 +237,15 @@ def resolve_conflicts(self, new_chain_list) -> tuple[bool, list]: temp_difficulty = new_chain_list[0].difficulty temp_avg_block_time = self.target_block_time - # Verify and apply blocks 1 to N + # Verify and apply blocks 1 to N through the shared pipeline for i in range(1, len(new_chain_list)): - prev_block = new_chain_list[i-1] - block = new_chain_list[i] - - if block.difficulty != temp_difficulty: - logger.warning("Reorg failed at block %s: Invalid difficulty. Expected %s, got %s", block.index, temp_difficulty, block.difficulty) - return False, [] - - try: - validate_block_link_and_hash(prev_block, block) - except ValueError as exc: - logger.warning("Reorg failed at block %s: %s", block.index, exc) - return False, [] - - receipts = [] - for tx in block.transactions: - from .validators import ValidationStatus - status, receipt = temp_state.validate_and_apply_with_status(tx) - if status != ValidationStatus.VALID: - logger.warning("Reorg failed: Transaction validation failed in block %s", block.index) - return False, [] - receipts.append(receipt) - - total_fees = sum(getattr(r, 'gas_used', 0) for r in receipts) - if block.miner: - temp_state.credit_mining_reward(block.miner, reward=temp_state.DEFAULT_MINING_REWARD + total_fees) - - computed_receipt_root = calculate_receipt_root(receipts) - if block.receipt_root != computed_receipt_root: - logger.warning("Reorg failed: Invalid receipt root at block %s. Expected %s, got %s", block.index, computed_receipt_root, block.receipt_root) - return False, [] - - if [r.to_dict() for r in block.receipts] != [r.to_dict() for r in receipts]: - logger.warning("Reorg failed: Receipt payload mismatch at block %s", block.index) - return False, [] - - if block.state_root != temp_state.state_root(): - logger.warning("Reorg failed: Invalid state root at block %s", block.index) + status, temp_difficulty, temp_avg_block_time = self._apply_block( + new_chain_list[i - 1], new_chain_list[i], temp_state, temp_difficulty, temp_avg_block_time + ) + if status != ValidationStatus.VALID: + logger.warning("Reorg failed at block %s", new_chain_list[i].index) return False, [] - # Update EMA difficulty state for reorg validation - time_diff = block.timestamp - prev_block.timestamp - temp_avg_block_time = self.alpha * time_diff + (1 - self.alpha) * temp_avg_block_time - if temp_avg_block_time > self.target_block_time: - temp_difficulty = max(1, temp_difficulty - 1) - elif temp_avg_block_time < self.target_block_time: - temp_difficulty += 1 - # 4. Success! Compute orphaned transactions. old_txs = {tx.tx_id: tx for b in original_chain[1:] for tx in b.transactions} new_tx_ids = {tx.tx_id for b in new_chain_list[1:] for tx in b.transactions} diff --git a/minichain/contract.py b/minichain/contract.py index 7aca420..9615038 100644 --- a/minichain/contract.py +++ b/minichain/contract.py @@ -19,7 +19,7 @@ def trace_calls(self, frame, event, arg): raise OutOfGasException("Out of gas!") return self.trace_calls -import json # Moved to module-level import +import json logger = logging.getLogger(__name__) def _safe_exec_worker(code, globals_dict, context_dict, result_queue, gas_limit): @@ -209,17 +209,12 @@ def _validate_code_ast(self, code): if isinstance(node, (ast.Import, ast.ImportFrom)): logger.warning("Rejected contract code with import statement.") return False - if isinstance(node, ast.Call): - if isinstance(node.func, ast.Name) and node.func.id == 'type': - logger.warning("Rejected type() call.") - return False - if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in {"getattr", "setattr", "delattr"}: + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in {"type", "getattr", "setattr", "delattr"}: logger.warning("Rejected direct call to %s.", node.func.id) return False - if isinstance(node, ast.Constant) and isinstance(node.value, str): - if "__" in node.value: - logger.warning("Rejected string literal with double-underscore.") - return False + if isinstance(node, ast.Constant) and isinstance(node.value, str) and "__" in node.value: + logger.warning("Rejected string literal with double-underscore.") + return False if isinstance(node, ast.JoinedStr): # f-strings logger.warning("Rejected f-string usage.") return False diff --git a/minichain/p2p.py b/minichain/p2p.py index 33cc9f7..b1e8712 100644 --- a/minichain/p2p.py +++ b/minichain/p2p.py @@ -21,6 +21,13 @@ logger = logging.getLogger(__name__) + +def _peer_id_from_addr(peer_addr: str) -> str: + """Strip the "peer:" prefix from a peer address, if present.""" + prefix = "peer:" + return peer_addr[len(prefix):] if peer_addr.startswith(prefix) else peer_addr + + SUPPORTED_MESSAGE_TYPES = {"hello", "tx", "block", "chain_request", "chain_response"} PROTOCOL_ID = TProtocol("/minichain/1.0.0") MAX_FRAME_BYTES = 1 * 1024 * 1024 # 1 MB @@ -53,25 +60,23 @@ def __init__( self._peer_count = 0 self._peer_count_lock = threading.Lock() - # Misbehavior tracking + # Misbehavior tracking, keyed directly by ValidationStatus so there is a + # single vocabulary for statuses (no parallel string keys to convert). self.data_path = data_path self.thresholds = { - "malformed": malformed_threshold, - "failed": failed_threshold, - "invalid": invalid_threshold, + ValidationStatus.MALFORMED: malformed_threshold, + ValidationStatus.FAILED: failed_threshold, + ValidationStatus.INVALID: invalid_threshold, } self.decay_interval_minutes = decay_interval_minutes - # { peer_id_str -> {"malformed": int, "failed": int, "invalid": int} } + # { peer_id_str -> { ValidationStatus -> int } } self._peer_counters: dict = {} if self.decay_interval_minutes <= 0: raise ValueError(f"decay_interval_minutes must be positive, got {self.decay_interval_minutes}") - if self.thresholds["malformed"] <= 0: - raise ValueError(f"malformed_threshold must be positive, got {self.thresholds['malformed']}") - if self.thresholds["failed"] <= 0: - raise ValueError(f"failed_threshold must be positive, got {self.thresholds['failed']}") - if self.thresholds["invalid"] <= 0: - raise ValueError(f"invalid_threshold must be positive, got {self.thresholds['invalid']}") + for status, value in self.thresholds.items(): + if value <= 0: + raise ValueError(f"{status.name.lower()}_threshold must be positive, got {value}") def register_handler(self, handler_callback): self._handler_callback = handler_callback @@ -102,14 +107,17 @@ def _message_id(self, msg_type, payload): if msg_type == "block": return payload["hash"] return None + def _seen_set(self, msg_type): + return self._seen_tx_ids if msg_type == "tx" else self._seen_block_hashes + def _is_duplicate(self, msg_type, payload): mid = self._message_id(msg_type, payload) - if not mid: return False - return mid in (self._seen_tx_ids if msg_type == "tx" else self._seen_block_hashes) + return bool(mid) and mid in self._seen_set(msg_type) def _mark_seen(self, msg_type, payload): mid = self._message_id(msg_type, payload) - if mid: (self._seen_tx_ids if msg_type == "tx" else self._seen_block_hashes).add(mid) + if mid: + self._seen_set(msg_type).add(mid) async def _broadcast_raw(self, payload: dict): self._to_trio.put(("BROADCAST", payload)) @@ -143,17 +151,15 @@ def peer_count(self) -> int: # ── misbehavior helpers ────────────────────────────────────────────────── - def _increment_counter(self, peer_id: str, category: str) -> bool: + def _increment_counter(self, peer_id: str, status: ValidationStatus) -> bool: """ - Increment the named counter (malformed/failed/invalid) for peer_id. - Returns True if any counter now meets or exceeds its threshold. + Increment peer_id's counter for the given ValidationStatus. + Returns True if that counter now meets or exceeds its threshold. Called only from the asyncio thread — no lock needed. """ - if peer_id not in self._peer_counters: - self._peer_counters[peer_id] = {"malformed": 0, "failed": 0, "invalid": 0} - self._peer_counters[peer_id][category] += 1 - counts = self._peer_counters[peer_id] - return counts[category] >= self.thresholds[category] + counts = self._peer_counters.setdefault(peer_id, {s: 0 for s in self.thresholds}) + counts[status] += 1 + return counts[status] >= self.thresholds[status] async def _handle_validation_status( self, peer_id: str, peer_addr: str, status: ValidationStatus @@ -164,21 +170,16 @@ async def _handle_validation_status( FAILED → drop silently; ban + disconnect if counter >= threshold INVALID → always ban + disconnect (threshold configurable, default=1) """ - category = { - ValidationStatus.MALFORMED: "malformed", - ValidationStatus.FAILED: "failed", - ValidationStatus.INVALID: "invalid", - }.get(status) - if category is None: + if status not in self.thresholds: return - exceeded = self._increment_counter(peer_id, category) + exceeded = self._increment_counter(peer_id, status) if exceeded: - ban_peer(peer_id, reason=f"{category}_threshold_exceeded", path=self.data_path) + ban_peer(peer_id, reason=f"{status.name.lower()}_threshold_exceeded", path=self.data_path) logger.warning( "Banned peer %s: %s threshold (%d) exceeded", - peer_id, category, self.thresholds[category], + peer_id, status.name.lower(), self.thresholds[status], ) always_disconnect = status in (ValidationStatus.MALFORMED, ValidationStatus.INVALID) @@ -216,9 +217,7 @@ async def _asyncio_reader(self): msg_type = data.get("type") payload = data.get("data") peer_addr = data.get("_peer_addr", "") - peer_id = ( - peer_addr[len("peer:"):] if peer_addr.startswith("peer:") else peer_addr - ) + peer_id = _peer_id_from_addr(peer_addr) if msg_type not in SUPPORTED_MESSAGE_TYPES: continue @@ -246,9 +245,7 @@ async def _asyncio_reader(self): elif msg[0] == "MALFORMED": # JSON parse failure signalled from the Trio thread. peer_addr = msg[1] - peer_id = ( - peer_addr[len("peer:"):] if peer_addr.startswith("peer:") else peer_addr - ) + peer_id = _peer_id_from_addr(peer_addr) await self._handle_validation_status(peer_id, peer_addr, ValidationStatus.MALFORMED) elif msg[0] == "PEER_CONNECTED": diff --git a/minichain/persistence.py b/minichain/persistence.py index 8de4148..dc989ba 100644 --- a/minichain/persistence.py +++ b/minichain/persistence.py @@ -14,10 +14,13 @@ from __future__ import annotations +import copy import json import logging import os import sqlite3 +import time +from contextlib import contextmanager from typing import Any from .block import Block @@ -48,7 +51,7 @@ def save(blockchain: Blockchain, path: str = ".") -> None: with blockchain._lock: chain_data = [block.to_dict() for block in blockchain.chain] - state_data = json.loads(json.dumps(blockchain.state.accounts)) + state_data = copy.deepcopy(blockchain.state.accounts) _save_snapshot_to_sqlite(db_path, {"chain": chain_data, "state": state_data}) @@ -214,10 +217,11 @@ def _save_snapshot_to_sqlite(db_path: str, snapshot: dict[str, Any]) -> None: def _load_snapshot_from_sqlite(db_path: str) -> dict[str, Any]: + invalid = f"Invalid SQLite persistence data in '{db_path}'" try: conn = _connect(db_path) except sqlite3.DatabaseError as exc: - raise ValueError(f"Invalid SQLite persistence data in '{db_path}'") from exc + raise ValueError(invalid) from exc try: _require_schema(conn) @@ -232,18 +236,18 @@ def _load_snapshot_from_sqlite(db_path: str) -> dict[str, Any]: ("chain_length",), ).fetchone() except sqlite3.DatabaseError as exc: - raise ValueError(f"Invalid SQLite persistence data in '{db_path}'") from exc + raise ValueError(invalid) from exc finally: conn.close() if chain_length_row is None: - raise ValueError(f"Invalid SQLite persistence data in '{db_path}'") + raise ValueError(invalid) try: expected_chain_length = int(chain_length_row["value"]) except (TypeError, ValueError) as exc: - raise ValueError(f"Invalid SQLite persistence data in '{db_path}'") from exc + raise ValueError(invalid) from exc if expected_chain_length != len(block_rows): - raise ValueError(f"Invalid SQLite persistence data in '{db_path}'") + raise ValueError(invalid) try: chain = [json.loads(row["block_json"]) for row in block_rows] @@ -261,62 +265,59 @@ def _load_snapshot_from_sqlite(db_path: str) -> dict[str, Any]: # Banned Peers (Track 1) # --------------------------------------------------------------------------- -import time def _ensure_banned_peers_table(conn: sqlite3.Connection) -> None: conn.execute( "CREATE TABLE IF NOT EXISTS banned_peers (peer_id TEXT PRIMARY KEY, reason TEXT, timestamp REAL)" ) -def ban_peer(peer_id: str, reason: str, path: str = ".") -> None: + +@contextmanager +def _banned_peers_conn(path: str, create: bool): + """ + Yield a connection with the banned_peers table ensured, always closing it. + When *create* is False and no DB exists yet, yields None so read-only callers + can short-circuit without touching the filesystem. + """ db_path = os.path.join(path, _DB_FILE) - os.makedirs(path, exist_ok=True) + if not create and not os.path.exists(db_path): + yield None + return + if create: + os.makedirs(path, exist_ok=True) conn = _connect(db_path) try: _ensure_banned_peers_table(conn) - with conn: - conn.execute( - "INSERT OR REPLACE INTO banned_peers (peer_id, reason, timestamp) VALUES (?, ?, ?)", - (peer_id, reason, time.time()) - ) + yield conn finally: conn.close() +def ban_peer(peer_id: str, reason: str, path: str = ".") -> None: + with _banned_peers_conn(path, create=True) as conn, conn: + conn.execute( + "INSERT OR REPLACE INTO banned_peers (peer_id, reason, timestamp) VALUES (?, ?, ?)", + (peer_id, reason, time.time()) + ) + def unban_peer(peer_id: str, path: str = ".") -> None: - db_path = os.path.join(path, _DB_FILE) - if not os.path.exists(db_path): - return - conn = _connect(db_path) - try: - _ensure_banned_peers_table(conn) + with _banned_peers_conn(path, create=False) as conn: + if conn is None: + return with conn: conn.execute("DELETE FROM banned_peers WHERE peer_id = ?", (peer_id,)) - finally: - conn.close() def is_peer_banned(peer_id: str, path: str = ".") -> bool: - db_path = os.path.join(path, _DB_FILE) - if not os.path.exists(db_path): - return False - conn = _connect(db_path) - try: - _ensure_banned_peers_table(conn) - row = conn.execute("SELECT peer_id FROM banned_peers WHERE peer_id = ?", (peer_id,)).fetchone() - return row is not None - finally: - conn.close() + with _banned_peers_conn(path, create=False) as conn: + if conn is None: + return False + return conn.execute("SELECT peer_id FROM banned_peers WHERE peer_id = ?", (peer_id,)).fetchone() is not None def get_banned_peers(path: str = ".") -> list[dict[str, Any]]: - db_path = os.path.join(path, _DB_FILE) - if not os.path.exists(db_path): - return [] - conn = _connect(db_path) - try: - _ensure_banned_peers_table(conn) + with _banned_peers_conn(path, create=False) as conn: + if conn is None: + return [] rows = conn.execute("SELECT peer_id, reason, timestamp FROM banned_peers ORDER BY timestamp DESC").fetchall() return [{"peer_id": r["peer_id"], "reason": r["reason"], "timestamp": r["timestamp"]} for r in rows] - finally: - conn.close() # --------------------------------------------------------------------------- diff --git a/minichain/rpc.py b/minichain/rpc.py index 58e4733..530c3aa 100644 --- a/minichain/rpc.py +++ b/minichain/rpc.py @@ -34,13 +34,8 @@ async def handle_rpc(self, request): return web.json_response({"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None}) if isinstance(req_data, list): - responses = [] - for req in req_data: - responses.append(await self._process_single(req)) - return web.json_response(responses) - else: - response = await self._process_single(req_data) - return web.json_response(response) + return web.json_response([await self._process_single(req) for req in req_data]) + return web.json_response(await self._process_single(req_data)) async def _process_single(self, req): if not isinstance(req, dict) or "method" not in req or req.get("jsonrpc") != "2.0": diff --git a/minichain/state.py b/minichain/state.py index bbe3ad6..92431dd 100644 --- a/minichain/state.py +++ b/minichain/state.py @@ -86,36 +86,28 @@ def restore(self, snapshot_data): """ self.accounts = copy.deepcopy(snapshot_data) - def validate_and_apply(self, tx): - """ - Validate and apply a transaction. - Returns: Receipt|None - """ - # Semantic validation: amount and fee must be non-negative integers + @staticmethod + def _amounts_well_formed(tx): + """Semantic guard: amount and fee must be non-negative integers.""" if not isinstance(tx.amount, int) or tx.amount < 0: - return None + return False fee = getattr(tx, "fee", 0) - if not isinstance(fee, int) or fee < 0: - return None - return self.apply_transaction(tx) + return isinstance(fee, int) and fee >= 0 def validate_and_apply_with_status(self, tx): """ Validate and apply a transaction, bubbling up the precise ValidationStatus. + This is the single core path; the other entry points delegate to it. Returns: (ValidationStatus, Receipt|None) """ from .validators import ValidationStatus - if not isinstance(tx.amount, int) or tx.amount < 0: - return ValidationStatus.MALFORMED, None - fee = getattr(tx, "fee", 0) - if not isinstance(fee, int) or fee < 0: + if not self._amounts_well_formed(tx): return ValidationStatus.MALFORMED, None status = self.verify_transaction_logic(tx) if status != ValidationStatus.VALID: return status, None - # verify_transaction_logic already passed — skip the second call inside apply_transaction. return ValidationStatus.VALID, self._apply_validated_tx(tx) def apply_transaction(self, tx): @@ -123,15 +115,10 @@ def apply_transaction(self, tx): Validates and applies a transaction. Returns: Receipt object if valid, None if invalid. """ - if not isinstance(tx.amount, int) or tx.amount < 0: - return None - fee = getattr(tx, "fee", 0) - if not isinstance(fee, int) or fee < 0: - return None - from .validators import ValidationStatus - if self.verify_transaction_logic(tx) != ValidationStatus.VALID: - return None - return self._apply_validated_tx(tx) + return self.validate_and_apply_with_status(tx)[1] + + # Backwards-compatible alias for the receipt-only entry point. + validate_and_apply = apply_transaction def _apply_validated_tx(self, tx): """ @@ -177,6 +164,11 @@ def _apply_validated_tx(self, tx): # Credit contract balance receiver['balance'] += tx.amount + # Undo the value transfer while keeping the nonce increment (used on failure paths). + def revert_transfer(): + receiver['balance'] -= tx.amount + sender['balance'] += tx.amount + result = self.contract_machine.execute( contract_address=tx.receiver, sender_address=tx.sender, @@ -191,18 +183,14 @@ def _apply_validated_tx(self, tx): sender['balance'] += gas_refund if not result.get("success"): - # Rollback transfer if execution fails, but keep nonce incremented - receiver['balance'] -= tx.amount - sender['balance'] += tx.amount # Refund amount + revert_transfer() return Receipt(tx.tx_id, status=0, error_message=result.get("error", "Execution failed"), gas_used=gas_used) transfers = result.get("transfers", []) total_transferred_out = sum(t["amount"] for t in transfers) if total_transferred_out > receiver['balance']: - # Rollback transfer if execution attempts to spend more than balance - receiver['balance'] -= tx.amount - sender['balance'] += tx.amount # Refund amount + revert_transfer() return Receipt(tx.tx_id, status=0, error_message="Insufficient contract balance for transfers", gas_used=gas_used) # Execution & transfers valid: commit state changes atomically From c97295b9466fbed776572f42162fb962489935cc Mon Sep 17 00:00:00 2001 From: siddhant Date: Mon, 6 Jul 2026 14:45:03 +0530 Subject: [PATCH 2/2] refactor codebase --- minichain/contract.py | 25 +++++++++++++++---------- minichain/rpc.py | 23 +++++++++++++++-------- 2 files changed, 30 insertions(+), 18 deletions(-) diff --git a/minichain/contract.py b/minichain/contract.py index 9615038..87799a7 100644 --- a/minichain/contract.py +++ b/minichain/contract.py @@ -99,6 +99,11 @@ class ContractMachine: def __init__(self, state): self.state = state + @staticmethod + def _fail(error, gas_used=0): + """Uniform failure result for execute().""" + return {"success": False, "gas_used": gas_used, "error": error} + def execute(self, contract_address, sender_address, payload, amount, gas_limit): """ Executes the contract code associated with the contract_address. @@ -107,7 +112,7 @@ def execute(self, contract_address, sender_address, payload, amount, gas_limit): account = self.state.get_account(contract_address) if not account: - return {"success": False, "gas_used": 0, "error": "Account not found"} + return self._fail("Account not found") code = account.get("code") @@ -115,11 +120,11 @@ def execute(self, contract_address, sender_address, payload, amount, gas_limit): storage = dict(account.get("storage", {})) if not code: - return {"success": False, "gas_used": 0, "error": "No code"} + return self._fail("No code") # AST Validation to prevent introspection if not self._validate_code_ast(code): - return {"success": False, "gas_used": 0, "error": "AST Validation Failed"} + return self._fail("AST Validation Failed") # Restricted builtins (explicit allowlist) safe_builtins = { @@ -131,7 +136,7 @@ def execute(self, contract_address, sender_address, payload, amount, gas_limit): "min": min, "max": max, "abs": abs, - "str": str, # Keeping str for basic functionality, relying on AST checks for safety + "str": str, # Keeping str for basic functionality, relying on AST checks for safety "bool": bool, "float": float, "list": list, @@ -170,30 +175,30 @@ def execute(self, contract_address, sender_address, payload, amount, gas_limit): p.kill() p.join() logger.error("Contract execution timed out") - return {"success": False, "gas_used": gas_limit, "error": "Execution timed out"} + return self._fail("Execution timed out", gas_limit) try: result = queue.get(timeout=1) except Exception: logger.error("Contract execution crashed without result") - return {"success": False, "gas_used": gas_limit, "error": "Crashed"} - + return self._fail("Crashed", gas_limit) + if result["status"] != "success": logger.error("Contract Execution Failed: %s", result.get('error')) - return {"success": False, "gas_used": result.get("gas_used", gas_limit), "error": result.get('error')} + return self._fail(result.get('error'), result.get("gas_used", gas_limit)) # Validate storage is JSON serializable try: json.dumps(result["storage"]) except (TypeError, ValueError): logger.error("Contract storage not JSON serializable") - return {"success": False, "gas_used": result.get("gas_used", gas_limit), "error": "Storage not JSON serializable"} + return self._fail("Storage not JSON serializable", result.get("gas_used", gas_limit)) return {"success": True, "gas_used": result["gas_used"], "transfers": result.get("transfers", []), "storage": result["storage"], "error": None} except Exception as e: logger.error("Contract Execution Failed", exc_info=True) - return {"success": False, "gas_used": gas_limit, "error": "System Error"} + return self._fail("System Error", gas_limit) def _validate_code_ast(self, code): """Reject code that uses double underscores or introspection.""" diff --git a/minichain/rpc.py b/minichain/rpc.py index 530c3aa..c2d8b2c 100644 --- a/minichain/rpc.py +++ b/minichain/rpc.py @@ -27,11 +27,19 @@ async def stop(self): if hasattr(self, 'runner'): await self.runner.cleanup() + @staticmethod + def _ok(result, req_id): + return {"jsonrpc": "2.0", "result": result, "id": req_id} + + @staticmethod + def _err(code, message, req_id): + return {"jsonrpc": "2.0", "error": {"code": code, "message": message}, "id": req_id} + async def handle_rpc(self, request): try: req_data = await request.json() except json.JSONDecodeError: - return web.json_response({"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None}) + return web.json_response(self._err(-32700, "Parse error", None)) if isinstance(req_data, list): return web.json_response([await self._process_single(req) for req in req_data]) @@ -39,8 +47,8 @@ async def handle_rpc(self, request): async def _process_single(self, req): if not isinstance(req, dict) or "method" not in req or req.get("jsonrpc") != "2.0": - return {"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request"}, "id": req.get("id") if isinstance(req, dict) else None} - + return self._err(-32600, "Invalid Request", req.get("id") if isinstance(req, dict) else None) + method = req["method"] params = req.get("params", []) req_id = req.get("id") @@ -65,8 +73,7 @@ async def _process_single(self, req): if not params: raise ValueError("Missing address") address = params[0] - account = self.chain.state.get_account(address) - result = account["balance"] if account else 0 + result = self.chain.state.get_account(address)["balance"] elif method == "mc_sendTransaction": if not params: raise ValueError("Missing transaction payload") @@ -80,9 +87,9 @@ async def _process_single(self, req): else: raise ValueError("Transaction rejected by Mempool") else: - return {"jsonrpc": "2.0", "error": {"code": -32601, "message": f"Method not found: {method}"}, "id": req_id} + return self._err(-32601, f"Method not found: {method}", req_id) - return {"jsonrpc": "2.0", "result": result, "id": req_id} + return self._ok(result, req_id) except Exception as e: logger.error("RPC Error processing %s: %s", method, e) - return {"jsonrpc": "2.0", "error": {"code": -32000, "message": str(e)}, "id": req_id} + return self._err(-32000, str(e), req_id)