""" Routes pour le vote et les bulletins. """ from fastapi import APIRouter, HTTPException, status, Depends, Request from sqlalchemy.orm import Session import base64 from .. import schemas, services from ..dependencies import get_db, get_current_voter from ..models import Voter from ..crypto.hashing import SecureHash router = APIRouter(prefix="/api/votes", tags=["votes"]) @router.post("/submit", response_model=schemas.VoteResponse) 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é """ # 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()) ) # Enregistrer le vote 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 ) # Marquer l'électeur comme ayant voté services.VoterService.mark_as_voted(db, current_voter.id) return schemas.VoteResponse( id=vote.id, ballot_hash=ballot_hash, timestamp=vote.timestamp ) @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: 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": "completed" if election.end_date < datetime.utcnow() else "active" }) return history