""" Routes pour le vote et les bulletins. """ import logging from fastapi import APIRouter, HTTPException, status, Depends, Request, Query from sqlalchemy.orm import Session from datetime import datetime, timezone import base64 import uuid from typing import Dict, Any, List from .. import schemas, services from ..dependencies import get_db, get_current_voter from ..models import Voter from ..crypto.hashing import SecureHash from ..blockchain import BlockchainManager from ..blockchain_client import BlockchainClient, get_blockchain_client_sync import asyncio logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/votes", tags=["votes"]) # Global blockchain manager instance (fallback for in-memory blockchain) blockchain_manager = BlockchainManager() # Global blockchain client instance for PoA validators blockchain_client: BlockchainClient = None def normalize_poa_blockchain_to_election_format(poa_data: Dict[str, Any], election_id: int) -> Dict[str, Any]: """ Normalize PoA blockchain format to election blockchain format. PoA format has nested transactions in each block. Election format has flat structure with transaction_id and encrypted_vote fields. Args: poa_data: Blockchain data from PoA validators election_id: Election ID for logging Returns: Blockchain data in election format """ logger.info(f"Normalizing PoA blockchain data for election {election_id}") normalized_blocks = [] # Convert each PoA block to election format for block in poa_data.get("blocks", []): logger.debug(f"Processing block {block.get('index')}: {len(block.get('transactions', []))} transactions") # If block has transactions (PoA format), convert each to a separate entry transactions = block.get("transactions", []) if len(transactions) == 0: # Genesis block or empty block - convert directly normalized_blocks.append({ "index": block.get("index"), "prev_hash": block.get("prev_hash", "0" * 64), "timestamp": block.get("timestamp", 0), "encrypted_vote": "", "transaction_id": "", "block_hash": block.get("block_hash", ""), "signature": block.get("signature", "") }) else: # Block with transactions - create one entry per transaction for tx in transactions: normalized_blocks.append({ "index": block.get("index"), "prev_hash": block.get("prev_hash", "0" * 64), "timestamp": block.get("timestamp", tx.get("timestamp", 0)), "encrypted_vote": tx.get("encrypted_vote", ""), "transaction_id": tx.get("voter_id", ""), # Use voter_id as transaction_id "block_hash": block.get("block_hash", ""), "signature": block.get("signature", "") }) logger.info(f"Normalized {len(poa_data.get('blocks', []))} PoA blocks to {len(normalized_blocks)} election format blocks") # Return in election format return { "blocks": normalized_blocks, "verification": poa_data.get("verification", { "chain_valid": True, "total_blocks": len(normalized_blocks), "total_votes": len(normalized_blocks) - 1 # Exclude genesis }) } async def init_blockchain_client(): """Initialize the blockchain client on startup""" global blockchain_client if blockchain_client is None: blockchain_client = BlockchainClient() await blockchain_client.refresh_validator_status() def get_blockchain_client() -> BlockchainClient: """Get the blockchain client instance""" global blockchain_client if blockchain_client is None: blockchain_client = BlockchainClient() return blockchain_client @router.post("") async def submit_simple_vote( vote_data: dict, current_voter: Voter = Depends(get_current_voter), db: Session = Depends(get_db), request: Request = None ): """ Soumettre un vote simple avec just élection_id et nom du candidat. Interface simplifiée pour l'application web. """ from .. import models election_id = vote_data.get('election_id') candidate_name = vote_data.get('choix') if not election_id or not candidate_name: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="election_id and choix are required" ) # Vérifier que l'électeur n'a pas déjà voté if services.VoteService.has_voter_voted(db, current_voter.id, election_id): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Voter has already voted in this election" ) # Vérifier que l'élection existe election = services.ElectionService.get_election(db, election_id) if not election: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Election not found" ) # Trouver le candidat par nom candidate = db.query(models.Candidate).filter( models.Candidate.name == candidate_name, models.Candidate.election_id == election_id ).first() if not candidate: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Candidate not found" ) # Enregistrer le vote (sans chiffrement pour l'MVP) import time from ..crypto.hashing import SecureHash ballot_hash = SecureHash.hash_bulletin( vote_id=current_voter.id, candidate_id=candidate.id, timestamp=int(time.time()) ) vote = services.VoteService.record_vote( db=db, voter_id=current_voter.id, election_id=election_id, candidate_id=candidate.id, encrypted_vote=b"", # Empty for MVP ballot_hash=ballot_hash, ip_address=request.client.host if request else None ) # Generate transaction ID for blockchain import uuid transaction_id = f"tx-{uuid.uuid4().hex[:12]}" # Submit vote to PoA blockchain blockchain_client = get_blockchain_client() await blockchain_client.refresh_validator_status() blockchain_response = { "status": "pending" } try: async with BlockchainClient() as poa_client: # Submit vote to PoA network submission_result = await poa_client.submit_vote( voter_id=current_voter.id, election_id=election_id, encrypted_vote="", # Empty for MVP (not encrypted) ballot_hash=ballot_hash, transaction_id=transaction_id ) blockchain_response = { "status": "submitted", "transaction_id": transaction_id, "block_hash": submission_result.get("block_hash"), "validator": submission_result.get("validator") } logger.info( f"Vote submitted to PoA: voter={current_voter.id}, " f"election={election_id}, tx={transaction_id}" ) except Exception as e: # Fallback: Record in local blockchain import traceback logger.warning(f"PoA submission failed: {e}") logger.warning(f"Exception type: {type(e).__name__}") logger.warning(f"Traceback: {traceback.format_exc()}") logger.warning("Falling back to local blockchain.") try: blockchain = blockchain_manager.get_or_create_blockchain(election_id) block = blockchain.add_block( encrypted_vote=ballot_hash, transaction_id=transaction_id ) blockchain_response = { "status": "submitted_fallback", "transaction_id": transaction_id, "block_index": block.index, "warning": "Vote recorded in local blockchain (PoA validators unreachable)" } except Exception as fallback_error: logger.error(f"Fallback blockchain also failed: {fallback_error}") blockchain_response = { "status": "database_only", "transaction_id": transaction_id, "warning": "Vote recorded in database but blockchain submission failed" } # Mark voter as having voted (only after confirming vote is recorded) # This ensures transactional consistency between database and marked status try: services.VoterService.mark_as_voted(db, current_voter.id) marked_as_voted = True except Exception as mark_error: logger.error(f"Failed to mark voter as voted: {mark_error}") # Note: Vote is already recorded, this is a secondary operation marked_as_voted = False return { "message": "Vote recorded successfully", "id": vote.id, "ballot_hash": ballot_hash, "timestamp": vote.timestamp, "blockchain": blockchain_response, "voter_marked_voted": marked_as_voted } @router.post("/submit") async def submit_vote( vote_bulletin: schemas.VoteBulletin, current_voter: Voter = Depends(get_current_voter), db: Session = Depends(get_db), request: Request = None ): """ Soumettre un vote chiffré via PoA blockchain. Le vote doit être: - Chiffré avec ElGamal - Accompagné d'une preuve ZK de validité Le vote est enregistré dans la PoA blockchain pour l'immuabilité. """ # Vérifier que l'électeur n'a pas déjà voté if services.VoteService.has_voter_voted( db, current_voter.id, vote_bulletin.election_id ): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Voter has already voted in this election" ) # Vérifier que l'élection existe election = services.ElectionService.get_election( db, vote_bulletin.election_id ) if not election: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Election not found" ) # Vérifier que le candidat existe from ..models import Candidate candidate = db.query(Candidate).filter( Candidate.id == vote_bulletin.candidate_id, Candidate.election_id == vote_bulletin.election_id ).first() if not candidate: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Candidate not found" ) # Décoder le vote chiffré try: encrypted_vote_bytes = base64.b64decode(vote_bulletin.encrypted_vote) except Exception: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid encrypted vote format" ) # Générer le hash du bulletin import time ballot_hash = SecureHash.hash_bulletin( vote_id=current_voter.id, candidate_id=vote_bulletin.candidate_id, timestamp=int(time.time()) ) # Générer ID unique pour la blockchain (anonyme) transaction_id = f"tx-{uuid.uuid4().hex[:12]}" # Enregistrer le vote en base de données vote = services.VoteService.record_vote( db=db, voter_id=current_voter.id, election_id=vote_bulletin.election_id, candidate_id=vote_bulletin.candidate_id, encrypted_vote=encrypted_vote_bytes, ballot_hash=ballot_hash, ip_address=request.client.host if request else None ) # Soumettre le vote aux validateurs PoA blockchain_client = get_blockchain_client() await blockchain_client.refresh_validator_status() blockchain_status = "pending" marked_as_voted = False try: async with BlockchainClient() as poa_client: # Soumettre le vote au réseau PoA submission_result = await poa_client.submit_vote( voter_id=current_voter.id, election_id=vote_bulletin.election_id, encrypted_vote=vote_bulletin.encrypted_vote, ballot_hash=ballot_hash, transaction_id=transaction_id ) blockchain_status = "submitted" logger.info( f"Vote submitted to PoA: voter={current_voter.id}, " f"election={vote_bulletin.election_id}, tx={transaction_id}" ) except Exception as e: # Fallback: Try to record in local blockchain logger.warning(f"PoA submission failed: {e}. Falling back to local blockchain.") try: blockchain = blockchain_manager.get_or_create_blockchain(vote_bulletin.election_id) block = blockchain.add_block( encrypted_vote=vote_bulletin.encrypted_vote, transaction_id=transaction_id ) blockchain_status = "submitted_fallback" except Exception as fallback_error: logger.error(f"Fallback blockchain also failed: {fallback_error}") blockchain_status = "database_only" # Mark voter as having voted (only after vote is confirmed recorded) # This ensures consistency regardless of blockchain status try: services.VoterService.mark_as_voted(db, current_voter.id) marked_as_voted = True except Exception as mark_error: logger.error(f"Failed to mark voter as voted: {mark_error}") # Note: Vote is already recorded, this is a secondary operation marked_as_voted = False return { "id": vote.id, "transaction_id": transaction_id, "ballot_hash": ballot_hash, "timestamp": vote.timestamp, "status": blockchain_status, "voter_marked_voted": marked_as_voted } @router.get("/status") def get_vote_status( election_id: int, current_voter: Voter = Depends(get_current_voter), db: Session = Depends(get_db) ): """Vérifier si l'électeur a déjà voté pour une élection""" has_voted = services.VoteService.has_voter_voted( db, current_voter.id, election_id ) return {"has_voted": has_voted} @router.get("/history", response_model=list) def get_voter_history( current_voter: Voter = Depends(get_current_voter), db: Session = Depends(get_db) ): """Récupérer l'historique des votes de l'électeur actuel""" from .. import models from datetime import datetime, timezone votes = db.query(models.Vote).filter( models.Vote.voter_id == current_voter.id ).all() # Retourner la structure avec infos des élections history = [] now = datetime.now(timezone.utc) for vote in votes: election = db.query(models.Election).filter( models.Election.id == vote.election_id ).first() candidate = db.query(models.Candidate).filter( models.Candidate.id == vote.candidate_id ).first() if election: # Ensure dates are timezone-aware for comparison start_date = election.start_date end_date = election.end_date # Make naive datetimes aware if needed if start_date and start_date.tzinfo is None: start_date = start_date.replace(tzinfo=timezone.utc) if end_date and end_date.tzinfo is None: end_date = end_date.replace(tzinfo=timezone.utc) # Déterminer le statut de l'élection if start_date and start_date > now: status_val = "upcoming" elif end_date and end_date < now: status_val = "closed" else: status_val = "active" history.append({ "vote_id": vote.id, "election_id": election.id, "election_name": election.name, "candidate_name": candidate.name if candidate else "Unknown", "candidate_id": vote.candidate_id, "candidate_id": candidate.id if candidate else None, "vote_date": vote.timestamp, "election_status": status_val }) return history @router.post("/setup") async def setup_election( election_id: int, current_voter: Voter = Depends(get_current_voter), db: Session = Depends(get_db) ): """ Initialiser une élection avec les clés cryptographiques. Crée une blockchain pour l'élection et génère les clés publiques pour le chiffrement ElGamal côté client. """ from .. import models from ..crypto.encryption import ElGamalEncryption # Vérifier que l'élection existe election = services.ElectionService.get_election(db, election_id) if not election: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Election not found" ) # Générer ou récupérer la blockchain pour cette élection blockchain = blockchain_manager.get_or_create_blockchain(election_id) # Générer les clés ElGamal si nécessaire if not election.public_key: elgamal = ElGamalEncryption() # Store as base64-encoded bytes (database column is LargeBinary) # public_key_bytes returns UTF-8 "p:g:h", then encode to base64 election.public_key = base64.b64encode(elgamal.public_key_bytes) db.add(election) db.commit() # Ensure public key is base64-encoded for client pubkey_to_return = election.public_key if isinstance(pubkey_to_return, bytes): pubkey_str = pubkey_to_return.decode('utf-8') # If it's plain "p:g:h" format, encode it to base64 if ':' in pubkey_str and not pubkey_str.startswith('MjM6'): # Not already base64 pubkey_to_return = base64.b64encode(pubkey_str.encode('utf-8')).decode('ascii') else: # Already base64, just decode to string pubkey_to_return = pubkey_str return { "status": "initialized", "election_id": election_id, "public_keys": { "elgamal_pubkey": pubkey_to_return }, "blockchain_blocks": blockchain.get_block_count() } @router.get("/public-keys") async def get_public_keys( election_id: int = Query(...), db: Session = Depends(get_db) ): """ Récupérer les clés publiques pour le chiffrement côté client. Accessible sans authentification pour permettre le chiffrement avant la connexion (si applicable). """ from .. import models # Vérifier que l'élection existe election = db.query(models.Election).filter( models.Election.id == election_id ).first() if not election: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Election not found" ) if not election.public_key: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Election keys not initialized. Call /setup first." ) # Ensure public key is base64-encoded for client pubkey_to_return = election.public_key if isinstance(pubkey_to_return, bytes): pubkey_str = pubkey_to_return.decode('utf-8') # If it's plain "p:g:h" format, encode it to base64 if ':' in pubkey_str and not pubkey_str.startswith('MjM6'): # Not already base64 pubkey_to_return = base64.b64encode(pubkey_str.encode('utf-8')).decode('ascii') else: # Already base64, just decode to string pubkey_to_return = pubkey_str return { "elgamal_pubkey": pubkey_to_return } @router.get("/blockchain") async def get_blockchain( election_id: int = Query(...), db: Session = Depends(get_db) ): """ Récupérer l'état complet de la blockchain pour une élection. Retourne tous les blocs et l'état de vérification. Requête d'abord aux validateurs PoA, puis fallback sur blockchain locale. """ # Vérifier que l'élection existe election = services.ElectionService.get_election(db, election_id) if not election: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Election not found" ) # Try to get blockchain state from PoA validators first try: async with BlockchainClient() as poa_client: blockchain_data = await poa_client.get_blockchain_state(election_id) if blockchain_data: logger.info(f"Got blockchain state from PoA for election {election_id}") # Normalize PoA format to election blockchain format return normalize_poa_blockchain_to_election_format(blockchain_data, election_id) except Exception as e: logger.warning(f"Failed to get blockchain from PoA: {e}") # Fallback to local blockchain manager logger.info(f"Falling back to local blockchain for election {election_id}") blockchain = blockchain_manager.get_or_create_blockchain(election_id) return blockchain.get_blockchain_data() @router.get("/results") async def get_results( election_id: int = Query(...), db: Session = Depends(get_db) ): """ Obtenir les résultats comptabilisés d'une élection. Requête d'abord aux validateurs PoA, puis fallback sur blockchain locale. """ from .. import models # Vérifier que l'élection existe election = db.query(models.Election).filter( models.Election.id == election_id ).first() if not election: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Election not found" ) # Essayer d'obtenir les résultats du réseau PoA en premier try: async with BlockchainClient() as poa_client: poa_results = await poa_client.get_election_results(election_id) if poa_results: logger.info(f"Retrieved results from PoA validators for election {election_id}") return poa_results except Exception as e: logger.warning(f"Failed to get results from PoA: {e}") # Fallback: Utiliser la blockchain locale logger.info(f"Falling back to local blockchain for election {election_id}") # Compter les votes par candidat (simple pour MVP) votes = db.query(models.Vote).filter( models.Vote.election_id == election_id ).all() # Grouper par candidat vote_counts = {} for vote in votes: candidate = db.query(models.Candidate).filter( models.Candidate.id == vote.candidate_id ).first() if candidate: if candidate.name not in vote_counts: vote_counts[candidate.name] = 0 vote_counts[candidate.name] += 1 # Obtenir la blockchain blockchain = blockchain_manager.get_or_create_blockchain(election_id) total_votes = blockchain.get_vote_count() results = [] for candidate_name, count in vote_counts.items(): percentage = (count / total_votes * 100) if total_votes > 0 else 0 results.append({ "candidate_name": candidate_name, "vote_count": count, "percentage": round(percentage, 2) }) return { "election_id": election_id, "election_name": election.name, "total_votes": total_votes, "results": sorted(results, key=lambda x: x["vote_count"], reverse=True), "verification": { "chain_valid": blockchain.verify_chain_integrity(), "timestamp": datetime.now(timezone.utc).isoformat() } } @router.post("/verify-blockchain") async def verify_blockchain( election_id: int, db: Session = Depends(get_db) ): """ Vérifier l'intégrité de la blockchain pour une élection. Requête d'abord aux validateurs PoA, puis fallback sur blockchain locale. Vérifie: - La chaîne de hachage (chaque bloc lie au précédent) - Les signatures des blocs - L'absence de modification """ # Vérifier que l'élection existe election = services.ElectionService.get_election(db, election_id) if not election: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Election not found" ) # Essayer de vérifier sur les validateurs PoA en premier try: async with BlockchainClient() as poa_client: is_valid = await poa_client.verify_blockchain_integrity(election_id) if is_valid is not None: blockchain_state = await poa_client.get_blockchain_state(election_id) logger.info(f"Blockchain verification from PoA validators for election {election_id}: {is_valid}") return { "election_id": election_id, "chain_valid": is_valid, "total_blocks": blockchain_state.get("verification", {}).get("total_blocks", 0) if blockchain_state else 0, "total_votes": blockchain_state.get("verification", {}).get("total_votes", 0) if blockchain_state else 0, "status": "valid" if is_valid else "invalid", "source": "poa_validators" } except Exception as e: logger.warning(f"Failed to verify blockchain on PoA: {e}") # Fallback: Vérifier sur la blockchain locale logger.info(f"Falling back to local blockchain verification for election {election_id}") blockchain = blockchain_manager.get_or_create_blockchain(election_id) is_valid = blockchain.verify_chain_integrity() return { "election_id": election_id, "chain_valid": is_valid, "total_blocks": blockchain.get_block_count(), "total_votes": blockchain.get_vote_count(), "status": "valid" if is_valid else "invalid", "source": "local_blockchain" } @router.get("/transaction-status") async def get_transaction_status( transaction_id: str = Query(...), election_id: int = Query(...), db: Session = Depends(get_db) ): """ Check the confirmation status of a vote on the PoA blockchain. Returns: - status: "pending" or "confirmed" - confirmed: boolean - block_number: block where vote is confirmed (if confirmed) - block_hash: hash of the block (if confirmed) """ # Vérifier que l'élection existe election = services.ElectionService.get_election(db, election_id) if not election: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Election not found" ) # Essayer de vérifier le statut sur PoA en premier try: async with BlockchainClient() as poa_client: status_info = await poa_client.get_vote_confirmation_status( transaction_id, election_id ) if status_info: logger.info(f"Transaction status from PoA: {transaction_id} = {status_info['status']}") return { **status_info, "source": "poa_validators" } except Exception as e: logger.warning(f"Failed to get transaction status from PoA: {e}") # Fallback: Check local blockchain logger.debug(f"Falling back to local blockchain for transaction {transaction_id}") return { "status": "unknown", "confirmed": False, "transaction_id": transaction_id, "source": "local_fallback" } @router.get("/check") async def check_voter_vote( election_id: int = Query(...), current_voter: Voter = Depends(get_current_voter), db: Session = Depends(get_db) ): """ Vérifier si le votant a déjà voté dans une élection spécifique. """ from .. import models # Vérifier si le votant a voté dans cette élection vote_exists = db.query(models.Vote).filter( models.Vote.voter_id == current_voter.id, models.Vote.election_id == election_id ).first() is not None return { "has_voted": vote_exists, "election_id": election_id, "voter_id": current_voter.id } from datetime import datetime