""" Routes pour le vote et les bulletins. """ from fastapi import APIRouter, HTTPException, status, Depends, Request, Query from sqlalchemy.orm import Session import base64 import uuid 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 router = APIRouter(prefix="/api/votes", tags=["votes"]) # Global blockchain manager instance blockchain_manager = BlockchainManager() @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 ) return { "message": "Vote recorded successfully", "id": vote.id, "ballot_hash": ballot_hash, "timestamp": vote.timestamp } @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é. Le vote doit être: - Chiffré avec ElGamal - Accompagné d'une preuve ZK de validité Le vote est enregistré dans la 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 ) # Ajouter le vote à la 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 ) # Marquer l'électeur comme ayant voté services.VoterService.mark_as_voted(db, current_voter.id) return { "id": vote.id, "transaction_id": transaction_id, "block_index": block.index, "ballot_hash": ballot_hash, "timestamp": vote.timestamp } except Exception as e: # Logging error but still return success (vote is recorded) print(f"Blockchain error: {e}") services.VoterService.mark_as_voted(db, current_voter.id) return { "id": vote.id, "transaction_id": transaction_id, "ballot_hash": ballot_hash, "timestamp": vote.timestamp, "warning": "Vote recorded but blockchain update failed" } @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 votes = db.query(models.Vote).filter( models.Vote.voter_id == current_voter.id ).all() # Retourner la structure avec infos des élections history = [] 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: # Déterminer le statut de l'élection if election.start_date > datetime.utcnow(): status_val = "upcoming" elif election.end_date < datetime.utcnow(): 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", "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 ElGamal # 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 = ElGamal() election.public_key = elgamal.public_key_bytes db.commit() return { "status": "initialized", "election_id": election_id, "public_keys": { "elgamal_pubkey": base64.b64encode(election.public_key).decode() if election.public_key else None }, "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." ) return { "elgamal_pubkey": base64.b64encode(election.public_key).decode() } @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. """ # 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" ) 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. Utilise la somme homomorphe des votes chiffrés sur la blockchain. """ 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" ) # 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.utcnow().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. 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" ) 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" } from datetime import datetime